// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package metadata_test

import (
	"context"
	"os"
	"path/filepath"
	"testing"
	"unsafe"

	"github.com/apache/arrow-go/v18/parquet"
	"github.com/apache/arrow-go/v18/parquet/file"
	"github.com/apache/arrow-go/v18/parquet/metadata"
	"github.com/apache/arrow-go/v18/parquet/schema"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func generateTableMetaData(schema *schema.Schema, props *parquet.WriterProperties, nrows int, statsInt, statsFloat metadata.EncodedStatistics) (*metadata.FileMetaData, error) {
	fbuilder := metadata.NewFileMetadataBuilder(schema, props, nil)
	rg1Builder := fbuilder.AppendRowGroup()
	// metadata
	// row group 1
	col1Builder := rg1Builder.NextColumnChunk()
	col2Builder := rg1Builder.NextColumnChunk()
	// column metadata
	dictEncodingStats := map[parquet.Encoding]int32{parquet.Encodings.RLEDict: 1}
	dataEncodingStats := map[parquet.Encoding]int32{parquet.Encodings.Plain: 1, parquet.Encodings.RLE: 1}
	statsInt.Signed = true
	col1Builder.SetStats(statsInt)
	statsFloat.Signed = true
	col2Builder.SetStats(statsFloat)

	col1Builder.Finish(metadata.ChunkMetaInfo{int64(nrows) / 2, 4, 0, 10, 512, 600}, true, false, metadata.EncodingStats{dictEncodingStats, dataEncodingStats})
	col2Builder.Finish(metadata.ChunkMetaInfo{int64(nrows) / 2, 24, 0, 30, 512, 600}, true, false, metadata.EncodingStats{dictEncodingStats, dataEncodingStats})

	rg1Builder.SetNumRows(nrows / 2)
	rg1Builder.Finish(1024, -1)

	// rowgroup2 metadata
	rg2Builder := fbuilder.AppendRowGroup()
	col1Builder = rg2Builder.NextColumnChunk()
	col2Builder = rg2Builder.NextColumnChunk()
	// column metadata
	col1Builder.SetStats(statsInt)
	col2Builder.SetStats(statsFloat)
	dictEncodingStats = make(map[parquet.Encoding]int32)
	col1Builder.Finish(metadata.ChunkMetaInfo{int64(nrows) / 2, 0 /*dictionary page offset*/, 0, 10, 512, 600}, false /* has dictionary */, false, metadata.EncodingStats{dictEncodingStats, dataEncodingStats})
	col2Builder.Finish(metadata.ChunkMetaInfo{int64(nrows) / 2, 16, 0, 26, 512, 600}, true, false, metadata.EncodingStats{dictEncodingStats, dataEncodingStats})

	rg2Builder.SetNumRows(nrows / 2)
	rg2Builder.Finish(1024, -1)

	return fbuilder.Finish()
}

func assertStatsSet(t *testing.T, m *metadata.ColumnChunkMetaData) {
	ok, err := m.StatsSet()
	assert.NoError(t, err)
	assert.True(t, ok)
}

func assertStats(t *testing.T, m *metadata.ColumnChunkMetaData) metadata.TypedStatistics {
	s, err := m.Statistics()
	assert.NoError(t, err)
	assert.NotNil(t, s)
	return s
}

func TestBuildAccess(t *testing.T) {
	props := parquet.NewWriterProperties(parquet.WithVersion(parquet.V2_LATEST))

	fields := schema.FieldList{
		schema.NewInt32Node("int_col", parquet.Repetitions.Required, -1),
		schema.NewFloat32Node("float_col", parquet.Repetitions.Required, -1),
	}
	root, err := schema.NewGroupNode("schema", parquet.Repetitions.Repeated, fields, -1)
	require.NoError(t, err)
	schema := schema.NewSchema(root)

	var (
		nrows      int64   = 1000
		intMin     int32   = 100
		intMax     int32   = 200
		floatMin   float32 = 100.100
		floatMax   float32 = 200.200
		statsInt   metadata.EncodedStatistics
		statsFloat metadata.EncodedStatistics
	)

	statsInt.SetNullCount(0).
		SetDistinctCount(nrows).
		SetMin((*(*[4]byte)(unsafe.Pointer(&intMin)))[:]).
		SetMax((*(*[4]byte)(unsafe.Pointer(&intMax)))[:])

	statsFloat.SetNullCount(0).
		SetDistinctCount(nrows).
		SetMin((*(*[4]byte)(unsafe.Pointer(&floatMin)))[:]).
		SetMax((*(*[4]byte)(unsafe.Pointer(&floatMax)))[:])

	faccessor, err := generateTableMetaData(schema, props, int(nrows), statsInt, statsFloat)
	require.NoError(t, err)
	serialized, err := faccessor.SerializeString(context.Background())
	assert.NoError(t, err)
	faccessorCopy, err := metadata.NewFileMetaData([]byte(serialized), nil)
	assert.NoError(t, err)

	for _, accessor := range []*metadata.FileMetaData{faccessor, faccessorCopy} {
		// file metadata
		assert.Equal(t, nrows, accessor.NumRows)
		assert.Len(t, accessor.RowGroups, 2)
		assert.EqualValues(t, parquet.V2_LATEST, accessor.Version())
		assert.Equal(t, parquet.DefaultCreatedBy, accessor.GetCreatedBy())
		assert.Equal(t, 3, accessor.NumSchemaElements())

		// row group 1 metadata
		rg1Access := accessor.RowGroup(0)
		assert.Equal(t, 2, rg1Access.NumColumns())
		assert.Equal(t, nrows/2, rg1Access.NumRows())
		assert.Equal(t, int64(1200), rg1Access.TotalByteSize())
		assert.Equal(t, int64(1024), rg1Access.TotalCompressedSize())

		rg1Col1, err := rg1Access.ColumnChunk(0)
		assert.NoError(t, err)
		assert.Equal(t, rg1Access.FileOffset(), rg1Col1.DictionaryPageOffset())

		rg1Col2, err := rg1Access.ColumnChunk(1)
		assert.NoError(t, err)
		assertStatsSet(t, rg1Col1)
		assertStatsSet(t, rg1Col2)
		assert.Equal(t, statsInt.Min, assertStats(t, rg1Col1).EncodeMin())
		assert.Equal(t, statsInt.Max, assertStats(t, rg1Col1).EncodeMax())
		assert.Equal(t, statsFloat.Min, assertStats(t, rg1Col2).EncodeMin())
		assert.Equal(t, statsFloat.Max, assertStats(t, rg1Col2).EncodeMax())
		assert.Zero(t, assertStats(t, rg1Col1).NullCount())
		assert.Zero(t, assertStats(t, rg1Col2).NullCount())
		assert.Equal(t, nrows, assertStats(t, rg1Col1).DistinctCount())
		assert.Equal(t, nrows, assertStats(t, rg1Col2).DistinctCount())
		assert.Equal(t, metadata.DefaultCompressionType, rg1Col1.Compression())
		assert.Equal(t, metadata.DefaultCompressionType, rg1Col2.Compression())
		assert.Equal(t, nrows/2, rg1Col1.NumValues())
		assert.Equal(t, nrows/2, rg1Col2.NumValues())
		assert.Len(t, rg1Col1.Encodings(), 3)
		assert.Len(t, rg1Col2.Encodings(), 3)
		assert.EqualValues(t, 512, rg1Col1.TotalCompressedSize())
		assert.EqualValues(t, 512, rg1Col2.TotalCompressedSize())
		assert.EqualValues(t, 600, rg1Col1.TotalUncompressedSize())
		assert.EqualValues(t, 600, rg1Col2.TotalUncompressedSize())
		assert.EqualValues(t, 4, rg1Col1.DictionaryPageOffset())
		assert.EqualValues(t, 24, rg1Col2.DictionaryPageOffset())
		assert.EqualValues(t, 10, rg1Col1.DataPageOffset())
		assert.EqualValues(t, 30, rg1Col2.DataPageOffset())
		assert.Len(t, rg1Col1.EncodingStats(), 3)
		assert.Len(t, rg1Col2.EncodingStats(), 3)

		// row group 2 metadata
		rg2Access := accessor.RowGroup(1)
		assert.Equal(t, 2, rg2Access.NumColumns())
		assert.Equal(t, nrows/2, rg2Access.NumRows())
		assert.EqualValues(t, 1200, rg2Access.TotalByteSize())
		assert.EqualValues(t, 1024, rg2Access.TotalCompressedSize())

		rg2Col1, err := rg2Access.ColumnChunk(0)
		assert.NoError(t, err)
		assert.Equal(t, rg2Access.FileOffset(), rg2Col1.DataPageOffset())

		rg2Col2, err := rg2Access.ColumnChunk(1)
		assert.NoError(t, err)
		assertStatsSet(t, rg1Col1)
		assertStatsSet(t, rg1Col2)
		assert.Equal(t, statsInt.Min, assertStats(t, rg1Col1).EncodeMin())
		assert.Equal(t, statsInt.Max, assertStats(t, rg1Col1).EncodeMax())
		assert.Equal(t, statsFloat.Min, assertStats(t, rg1Col2).EncodeMin())
		assert.Equal(t, statsFloat.Max, assertStats(t, rg1Col2).EncodeMax())
		assert.Zero(t, assertStats(t, rg1Col1).NullCount())
		assert.Zero(t, assertStats(t, rg1Col2).NullCount())
		assert.Equal(t, nrows, assertStats(t, rg1Col1).DistinctCount())
		assert.Equal(t, nrows, assertStats(t, rg1Col2).DistinctCount())
		assert.Equal(t, metadata.DefaultCompressionType, rg2Col1.Compression())
		assert.Equal(t, metadata.DefaultCompressionType, rg2Col2.Compression())
		assert.Equal(t, nrows/2, rg2Col1.NumValues())
		assert.Equal(t, nrows/2, rg2Col2.NumValues())
		assert.Len(t, rg2Col1.Encodings(), 2)
		assert.Len(t, rg2Col2.Encodings(), 3)
		assert.EqualValues(t, 512, rg2Col1.TotalCompressedSize())
		assert.EqualValues(t, 512, rg2Col2.TotalCompressedSize())
		assert.EqualValues(t, 600, rg2Col1.TotalUncompressedSize())
		assert.EqualValues(t, 600, rg2Col2.TotalUncompressedSize())
		assert.EqualValues(t, 0, rg2Col1.DictionaryPageOffset())
		assert.EqualValues(t, 16, rg2Col2.DictionaryPageOffset())
		assert.EqualValues(t, 10, rg2Col1.DataPageOffset())
		assert.EqualValues(t, 26, rg2Col2.DataPageOffset())
		assert.Len(t, rg2Col1.EncodingStats(), 2)
		assert.Len(t, rg2Col2.EncodingStats(), 2)

		assert.Empty(t, rg2Col1.FilePath())
		accessor.SetFilePath("/foo/bar/bar.parquet")
		assert.Equal(t, "/foo/bar/bar.parquet", rg2Col1.FilePath())
	}

	faccessor2, err := generateTableMetaData(schema, props, int(nrows), statsInt, statsFloat)
	require.NoError(t, err)
	faccessor.AppendRowGroups(faccessor2)
	assert.Len(t, faccessor.RowGroups, 4)
	assert.Equal(t, nrows*2, faccessor.NumRows)
	assert.EqualValues(t, parquet.V2_LATEST, faccessor.Version())
	assert.Equal(t, parquet.DefaultCreatedBy, faccessor.GetCreatedBy())
	assert.Equal(t, 3, faccessor.NumSchemaElements())

	faccessor1, err := faccessor.Subset([]int{2, 3})
	require.NoError(t, err)
	assert.True(t, faccessor1.Equals(faccessor2))

	faccessor1, err = faccessor2.Subset([]int{0})
	require.NoError(t, err)

	next, err := faccessor.Subset([]int{0})
	require.NoError(t, err)
	faccessor1.AppendRowGroups(next)

	sub, err := faccessor.Subset([]int{2, 0})
	require.NoError(t, err)
	assert.True(t, faccessor1.Equals(sub))
}

func TestV1VersionMetadata(t *testing.T) {
	props := parquet.NewWriterProperties(parquet.WithVersion(parquet.V1_0))

	fields := schema.FieldList{
		schema.NewInt32Node("int_col", parquet.Repetitions.Required, -1),
		schema.NewFloat32Node("float_col", parquet.Repetitions.Required, -1),
	}
	root, err := schema.NewGroupNode("schema", parquet.Repetitions.Repeated, fields, -1)
	require.NoError(t, err)
	schema := schema.NewSchema(root)

	fbuilder := metadata.NewFileMetadataBuilder(schema, props, nil)
	faccessor, err := fbuilder.Finish()
	require.NoError(t, err)
	assert.EqualValues(t, parquet.V1_0, faccessor.Version())
}

func TestKeyValueMetadata(t *testing.T) {
	props := parquet.NewWriterProperties(parquet.WithVersion(parquet.V1_0))

	fields := schema.FieldList{
		schema.NewInt32Node("int_col", parquet.Repetitions.Required, -1),
		schema.NewFloat32Node("float_col", parquet.Repetitions.Required, -1),
	}
	root, err := schema.NewGroupNode("schema", parquet.Repetitions.Repeated, fields, -1)
	require.NoError(t, err)
	schema := schema.NewSchema(root)
	kvmeta := metadata.NewKeyValueMetadata()
	kvmeta.Append("test_key", "test_value")

	fbuilder := metadata.NewFileMetadataBuilder(schema, props, kvmeta)
	faccessor, err := fbuilder.Finish()
	require.NoError(t, err)

	assert.True(t, faccessor.KeyValueMetadata().Equals(kvmeta))
}

func TestKeyValueMetadataAppend(t *testing.T) {
	props := parquet.NewWriterProperties(parquet.WithVersion(parquet.V1_0))

	fields := schema.FieldList{
		schema.NewInt32Node("int_col", parquet.Repetitions.Required, -1),
		schema.NewFloat32Node("float_col", parquet.Repetitions.Required, -1),
	}
	root, err := schema.NewGroupNode("schema", parquet.Repetitions.Repeated, fields, -1)
	require.NoError(t, err)
	schema := schema.NewSchema(root)

	kvmeta := metadata.NewKeyValueMetadata()
	key1 := "test_key1"
	value1 := "test_value1"
	require.NoError(t, kvmeta.Append(key1, value1))

	fbuilder := metadata.NewFileMetadataBuilder(schema, props, kvmeta)

	key2 := "test_key2"
	value2 := "test_value2"
	require.NoError(t, fbuilder.AppendKeyValueMetadata(key2, value2))
	faccessor, err := fbuilder.Finish()
	require.NoError(t, err)

	kv := faccessor.KeyValueMetadata()

	got1 := kv.FindValue(key1)
	require.NotNil(t, got1)
	assert.Equal(t, value1, *got1)

	got2 := kv.FindValue(key2)
	require.NotNil(t, got2)
	assert.Equal(t, value2, *got2)
}

func TestApplicationVersion(t *testing.T) {
	version := metadata.NewAppVersion("parquet-mr version 1.7.9")
	version1 := metadata.NewAppVersion("parquet-mr version 1.8.0")
	version2 := metadata.NewAppVersion("parquet-cpp version 1.0.0")
	version3 := metadata.NewAppVersion("")
	version4 := metadata.NewAppVersion("parquet-mr version 1.5.0ab-cdh5.5.0+cd (build abcd)")
	version5 := metadata.NewAppVersion("parquet-mr")

	assert.Equal(t, "parquet-mr", version.App)
	assert.Equal(t, 1, version.Version.Major)
	assert.Equal(t, 7, version.Version.Minor)
	assert.Equal(t, 9, version.Version.Patch)

	assert.Equal(t, "parquet-cpp", version2.App)
	assert.Equal(t, 1, version2.Version.Major)
	assert.Equal(t, 0, version2.Version.Minor)
	assert.Equal(t, 0, version2.Version.Patch)

	assert.Equal(t, "parquet-mr", version4.App)
	assert.Equal(t, "abcd", version4.Build)
	assert.Equal(t, 1, version4.Version.Major)
	assert.Equal(t, 5, version4.Version.Minor)
	assert.Equal(t, 0, version4.Version.Patch)
	assert.Equal(t, "ab", version4.Version.Unknown)
	assert.Equal(t, "cdh5.5.0", version4.Version.PreRelease)
	assert.Equal(t, "cd", version4.Version.BuildInfo)

	assert.Equal(t, "parquet-mr", version5.App)
	assert.Equal(t, 0, version5.Version.Major)
	assert.Equal(t, 0, version5.Version.Minor)
	assert.Equal(t, 0, version5.Version.Patch)

	assert.True(t, version.LessThan(version1))

	var stats metadata.EncodedStatistics
	assert.False(t, version1.HasCorrectStatistics(parquet.Types.Int96, schema.NoLogicalType{}, stats, schema.SortUNKNOWN))
	assert.True(t, version.HasCorrectStatistics(parquet.Types.Int32, schema.NoLogicalType{}, stats, schema.SortSIGNED))
	assert.False(t, version.HasCorrectStatistics(parquet.Types.ByteArray, schema.NoLogicalType{}, stats, schema.SortSIGNED))
	assert.True(t, version1.HasCorrectStatistics(parquet.Types.ByteArray, schema.NoLogicalType{}, stats, schema.SortSIGNED))
	assert.False(t, version1.HasCorrectStatistics(parquet.Types.ByteArray, schema.NoLogicalType{}, stats, schema.SortUNSIGNED))
	assert.True(t, version3.HasCorrectStatistics(parquet.Types.FixedLenByteArray, schema.NoLogicalType{}, stats, schema.SortSIGNED))

	// check that the old stats are correct if min and max are the same regardless of sort order
	var statsStr metadata.EncodedStatistics
	statsStr.SetMin([]byte("a")).SetMax([]byte("b"))
	assert.False(t, version1.HasCorrectStatistics(parquet.Types.ByteArray, schema.NoLogicalType{}, statsStr, schema.SortUNSIGNED))
	statsStr.SetMax([]byte("a"))
	assert.True(t, version1.HasCorrectStatistics(parquet.Types.ByteArray, schema.NoLogicalType{}, statsStr, schema.SortUNSIGNED))

	// check that the same holds true for ints
	var (
		intMin int32 = 100
		intMax int32 = 200
	)
	var statsInt metadata.EncodedStatistics
	statsInt.SetMin((*(*[4]byte)(unsafe.Pointer(&intMin)))[:])
	statsInt.SetMax((*(*[4]byte)(unsafe.Pointer(&intMax)))[:])
	assert.False(t, version1.HasCorrectStatistics(parquet.Types.ByteArray, schema.NoLogicalType{}, statsInt, schema.SortUNSIGNED))
	statsInt.SetMax((*(*[4]byte)(unsafe.Pointer(&intMin)))[:])
	assert.True(t, version1.HasCorrectStatistics(parquet.Types.ByteArray, schema.NoLogicalType{}, statsInt, schema.SortUNSIGNED))
}

func TestCheckBadDecimalStats(t *testing.T) {
	version1 := metadata.NewAppVersion("parquet-cpp version 3.0.0")
	version2 := metadata.NewAppVersion("parquet-cpp-arrow version 3.0.0")
	version3 := metadata.NewAppVersion("parquet-cpp-arrow version 4.0.0")

	var stats metadata.EncodedStatistics
	assert.False(t, version1.HasCorrectStatistics(parquet.Types.FixedLenByteArray, schema.NewDecimalLogicalType(5, 0), stats, schema.SortSIGNED))
	assert.False(t, version2.HasCorrectStatistics(parquet.Types.FixedLenByteArray, schema.NewDecimalLogicalType(5, 0), stats, schema.SortSIGNED))
	assert.True(t, version3.HasCorrectStatistics(parquet.Types.FixedLenByteArray, schema.NewDecimalLogicalType(5, 0), stats, schema.SortSIGNED))
}

func TestReadPageIndex(t *testing.T) {
	dir := os.Getenv("PARQUET_TEST_DATA")
	if dir == "" {
		t.Skip("PARQUET_TEST_DATA not set")
	}
	require.DirExists(t, dir)

	path := filepath.Join(dir, "alltypes_tiny_pages.parquet")
	rdr, err := file.OpenParquetFile(path, false)
	require.NoError(t, err)
	defer rdr.Close()

	fileMetadata := rdr.MetaData()
	assert.Equal(t, 1, fileMetadata.NumRowGroups())
	rgMeta := fileMetadata.RowGroup(0)
	assert.Equal(t, 13, rgMeta.NumColumns())

	ciOffsets := []int64{323583, 327502, 328009, 331928, 335847,
		339766, 350345, 354264, 364843, 384342, -1, 386473, 390392}
	ciLengths := []int64{3919, 507, 3919, 3919, 3919, 10579, 3919,
		10579, 19499, 2131, -1, 3919, 3919}
	oiOffsets := []int64{394311, 397814, 398637, 401888, 405139,
		408390, 413670, 416921, 422201, 431936, 435457, 446002, 449253}
	oiLengths := []int64{3503, 823, 3251, 3251, 3251, 5280, 3251, 5280,
		9735, 3521, 10545, 3251, 3251}

	for i := 0; i < rgMeta.NumColumns(); i++ {
		colMeta, err := rgMeta.ColumnChunk(i)
		require.NoError(t, err)

		ciLoc := colMeta.GetColumnIndexLocation()
		if i == 10 {
			assert.Nil(t, ciLoc)
		} else {
			assert.NotNil(t, ciLoc)
			assert.EqualValues(t, ciOffsets[i], ciLoc.Offset)
			assert.EqualValues(t, ciLengths[i], ciLoc.Length)
		}

		oiLoc := colMeta.GetOffsetIndexLocation()
		assert.NotNil(t, oiLoc)
		assert.EqualValues(t, oiOffsets[i], oiLoc.Offset)
		assert.EqualValues(t, oiLengths[i], oiLoc.Length)
	}
}
