mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-08-21 22:56:55 +00:00
Chris LuandGitHub
b4b0346f95
The worker assumed every file of a table sits under its catalog path, so loadFileByIcebergPath stripped the scheme off a recorded location and joined the remainder onto /buckets/<bucket>/<ns>/<table>. A table the REST catalog placed elsewhere in the bucket — which is what a client gets whenever the catalog path is already occupied — then resolves to a doubled path: lookup /buckets/lake/source/t/lake/source/t-0cd81bca-.../metadata/snap-.avro so the very first manifest list read fails and the job fails again on every scan interval, indefinitely. Resolve absolute references (s3:// URIs and /buckets paths) from the bucket root and keep relative ones under the table's own directory; the bucket-relative form is now the canonical key everywhere references are compared. That directory comes from the metadata location the catalog stores, so reads, writes and deletes all land where the table's other files are instead of splitting it across two trees. References outside the table's bucket are rejected rather than silently misresolved. Rewritten position-delete files now name their data file by absolute URI, the way the table itself names it, instead of a path relative to the table.
1416 lines
43 KiB
Go
1416 lines
43 KiB
Go
package iceberg
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"path"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/apache/iceberg-go"
|
|
"github.com/apache/iceberg-go/table"
|
|
"github.com/parquet-go/parquet-go"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
|
|
)
|
|
|
|
func TestParseConfig(t *testing.T) {
|
|
config := ParseConfig(nil)
|
|
|
|
if config.SnapshotRetentionHours != defaultSnapshotRetentionHours {
|
|
t.Errorf("expected SnapshotRetentionHours=%d, got %d", defaultSnapshotRetentionHours, config.SnapshotRetentionHours)
|
|
}
|
|
if config.MaxSnapshotsToKeep != defaultMaxSnapshotsToKeep {
|
|
t.Errorf("expected MaxSnapshotsToKeep=%d, got %d", defaultMaxSnapshotsToKeep, config.MaxSnapshotsToKeep)
|
|
}
|
|
if config.OrphanOlderThanHours != defaultOrphanOlderThanHours {
|
|
t.Errorf("expected OrphanOlderThanHours=%d, got %d", defaultOrphanOlderThanHours, config.OrphanOlderThanHours)
|
|
}
|
|
if config.MaxCommitRetries != defaultMaxCommitRetries {
|
|
t.Errorf("expected MaxCommitRetries=%d, got %d", defaultMaxCommitRetries, config.MaxCommitRetries)
|
|
}
|
|
if config.Operations != defaultOperations {
|
|
t.Errorf("expected Operations=%q, got %q", defaultOperations, config.Operations)
|
|
}
|
|
if config.RewriteStrategy != defaultRewriteStrategy {
|
|
t.Errorf("expected RewriteStrategy=%q, got %q", defaultRewriteStrategy, config.RewriteStrategy)
|
|
}
|
|
if config.SortMaxInputBytes != 0 {
|
|
t.Errorf("expected SortMaxInputBytes=0, got %d", config.SortMaxInputBytes)
|
|
}
|
|
}
|
|
|
|
func TestParseOperations(t *testing.T) {
|
|
tests := []struct {
|
|
input string
|
|
expected []string
|
|
wantErr bool
|
|
}{
|
|
{"all", []string{"compact", "rewrite_position_delete_files", "expire_snapshots", "remove_orphans", "rewrite_manifests"}, false},
|
|
{"", []string{"compact", "rewrite_position_delete_files", "expire_snapshots", "remove_orphans", "rewrite_manifests"}, false},
|
|
{"expire_snapshots", []string{"expire_snapshots"}, false},
|
|
{"compact", []string{"compact"}, false},
|
|
{"rewrite_position_delete_files", []string{"rewrite_position_delete_files"}, false},
|
|
{"rewrite_manifests,expire_snapshots", []string{"expire_snapshots", "rewrite_manifests"}, false},
|
|
{"compact,expire_snapshots", []string{"compact", "expire_snapshots"}, false},
|
|
{"remove_orphans, rewrite_manifests", []string{"remove_orphans", "rewrite_manifests"}, false},
|
|
{"expire_snapshots,remove_orphans,rewrite_manifests", []string{"expire_snapshots", "remove_orphans", "rewrite_manifests"}, false},
|
|
{"compact,expire_snapshots,remove_orphans,rewrite_manifests", []string{"compact", "expire_snapshots", "remove_orphans", "rewrite_manifests"}, false},
|
|
{"compact,rewrite_position_delete_files,rewrite_manifests", []string{"compact", "rewrite_position_delete_files", "rewrite_manifests"}, false},
|
|
{"unknown_op", nil, true},
|
|
{"expire_snapshots,bad_op", nil, true},
|
|
}
|
|
|
|
for _, tc := range tests {
|
|
result, err := parseOperations(tc.input)
|
|
if tc.wantErr {
|
|
if err == nil {
|
|
t.Errorf("parseOperations(%q) expected error, got %v", tc.input, result)
|
|
}
|
|
continue
|
|
}
|
|
if err != nil {
|
|
t.Errorf("parseOperations(%q) unexpected error: %v", tc.input, err)
|
|
continue
|
|
}
|
|
if len(result) != len(tc.expected) {
|
|
t.Errorf("parseOperations(%q) = %v, want %v", tc.input, result, tc.expected)
|
|
continue
|
|
}
|
|
for i := range result {
|
|
if result[i] != tc.expected[i] {
|
|
t.Errorf("parseOperations(%q)[%d] = %q, want %q", tc.input, i, result[i], tc.expected[i])
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestExtractMetadataVersion(t *testing.T) {
|
|
tests := []struct {
|
|
input string
|
|
expected int
|
|
}{
|
|
{"v1.metadata.json", 1},
|
|
{"v5.metadata.json", 5},
|
|
{"v100.metadata.json", 100},
|
|
{"v0.metadata.json", 0},
|
|
{"invalid.metadata.json", 0},
|
|
{"metadata.json", 0},
|
|
{"", 0},
|
|
{"v.metadata.json", 0},
|
|
{"v7-1709766000.metadata.json", 7},
|
|
{"v42-abc123.metadata.json", 42},
|
|
{"v5-.metadata.json", 5},
|
|
}
|
|
|
|
for _, tc := range tests {
|
|
result := extractMetadataVersion(tc.input)
|
|
if result != tc.expected {
|
|
t.Errorf("extractMetadataVersion(%q) = %d, want %d", tc.input, result, tc.expected)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestNeedsMaintenanceNoSnapshots(t *testing.T) {
|
|
config := Config{
|
|
SnapshotRetentionHours: 24,
|
|
MaxSnapshotsToKeep: 2,
|
|
}
|
|
|
|
meta := buildTestMetadata(t, nil)
|
|
if needsMaintenance(meta, config) {
|
|
t.Error("expected no maintenance for table with no snapshots")
|
|
}
|
|
}
|
|
|
|
func TestNeedsMaintenanceExceedsMaxSnapshots(t *testing.T) {
|
|
config := Config{
|
|
SnapshotRetentionHours: 24 * 365, // very long retention
|
|
MaxSnapshotsToKeep: 2,
|
|
}
|
|
|
|
now := time.Now().UnixMilli()
|
|
snapshots := []table.Snapshot{
|
|
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
|
|
{SnapshotID: 2, TimestampMs: now + 1, ManifestList: "metadata/snap-2.avro"},
|
|
{SnapshotID: 3, TimestampMs: now + 2, ManifestList: "metadata/snap-3.avro"},
|
|
}
|
|
meta := buildTestMetadata(t, snapshots)
|
|
if !needsMaintenance(meta, config) {
|
|
t.Error("expected maintenance for table exceeding max snapshots")
|
|
}
|
|
}
|
|
|
|
func TestNeedsMaintenanceWithinLimits(t *testing.T) {
|
|
config := Config{
|
|
SnapshotRetentionHours: 24 * 365, // very long retention
|
|
MaxSnapshotsToKeep: 5,
|
|
}
|
|
|
|
now := time.Now().UnixMilli()
|
|
snapshots := []table.Snapshot{
|
|
{SnapshotID: 1, TimestampMs: now, ManifestList: "metadata/snap-1.avro"},
|
|
}
|
|
meta := buildTestMetadata(t, snapshots)
|
|
if needsMaintenance(meta, config) {
|
|
t.Error("expected no maintenance for table within limits")
|
|
}
|
|
}
|
|
|
|
func TestNeedsMaintenanceOldSnapshot(t *testing.T) {
|
|
// Use a retention of 0 hours so that any snapshot is considered "old"
|
|
config := Config{
|
|
SnapshotRetentionHours: 0, // instant expiry
|
|
MaxSnapshotsToKeep: 10,
|
|
}
|
|
|
|
now := time.Now().UnixMilli()
|
|
snapshots := []table.Snapshot{
|
|
{SnapshotID: 1, TimestampMs: now - 1, ManifestList: "metadata/snap-1.avro"},
|
|
}
|
|
meta := buildTestMetadata(t, snapshots)
|
|
// With 0 retention, any snapshot with timestamp < now should need maintenance
|
|
if !needsMaintenance(meta, config) {
|
|
t.Error("expected maintenance for table with expired snapshot")
|
|
}
|
|
}
|
|
|
|
func TestCapabilityAndDescriptor(t *testing.T) {
|
|
handler := NewHandler(nil)
|
|
|
|
cap := handler.Capability()
|
|
if cap.JobType != jobType {
|
|
t.Errorf("expected job type %q, got %q", jobType, cap.JobType)
|
|
}
|
|
if !cap.CanDetect {
|
|
t.Error("expected CanDetect=true")
|
|
}
|
|
if !cap.CanExecute {
|
|
t.Error("expected CanExecute=true")
|
|
}
|
|
|
|
desc := handler.Descriptor()
|
|
if desc.JobType != jobType {
|
|
t.Errorf("expected job type %q, got %q", jobType, desc.JobType)
|
|
}
|
|
if desc.AdminConfigForm == nil {
|
|
t.Error("expected admin config form")
|
|
}
|
|
if desc.WorkerConfigForm == nil {
|
|
t.Error("expected worker config form")
|
|
}
|
|
if desc.AdminRuntimeDefaults == nil {
|
|
t.Error("expected admin runtime defaults")
|
|
}
|
|
if desc.AdminRuntimeDefaults.Enabled {
|
|
t.Error("expected disabled by default")
|
|
}
|
|
}
|
|
|
|
func TestBuildMaintenanceProposal(t *testing.T) {
|
|
handler := NewHandler(nil)
|
|
|
|
now := time.Now().UnixMilli()
|
|
snapshots := []table.Snapshot{
|
|
{SnapshotID: 1, TimestampMs: now},
|
|
{SnapshotID: 2, TimestampMs: now + 1},
|
|
}
|
|
meta := buildTestMetadata(t, snapshots)
|
|
|
|
info := tableInfo{
|
|
BucketName: "my-bucket",
|
|
Namespace: "analytics",
|
|
TableName: "events",
|
|
TablePath: "analytics/events",
|
|
Metadata: meta,
|
|
}
|
|
|
|
proposal := handler.buildMaintenanceProposal(info, "localhost:8888", "my-bucket")
|
|
|
|
expectedDedupe := "iceberg_maintenance:my-bucket/analytics/events"
|
|
if proposal.DedupeKey != expectedDedupe {
|
|
t.Errorf("expected dedupe key %q, got %q", expectedDedupe, proposal.DedupeKey)
|
|
}
|
|
if proposal.JobType != jobType {
|
|
t.Errorf("expected job type %q, got %q", jobType, proposal.JobType)
|
|
}
|
|
|
|
if readStringConfig(proposal.Parameters, "bucket_name", "") != "my-bucket" {
|
|
t.Error("expected bucket_name=my-bucket in parameters")
|
|
}
|
|
if readStringConfig(proposal.Parameters, "namespace", "") != "analytics" {
|
|
t.Error("expected namespace=analytics in parameters")
|
|
}
|
|
if readStringConfig(proposal.Parameters, "table_name", "") != "events" {
|
|
t.Error("expected table_name=events in parameters")
|
|
}
|
|
if readStringConfig(proposal.Parameters, "filer_address", "") != "localhost:8888" {
|
|
t.Error("expected filer_address=localhost:8888 in parameters")
|
|
}
|
|
if readStringConfig(proposal.Parameters, "resource_group", "") != "my-bucket" {
|
|
t.Error("expected resource_group=my-bucket in parameters")
|
|
}
|
|
if proposal.Labels["resource_group"] != "my-bucket" {
|
|
t.Error("expected resource_group label to be set")
|
|
}
|
|
}
|
|
|
|
func TestReadResourceGroupConfig(t *testing.T) {
|
|
cfg, err := readResourceGroupConfig(nil)
|
|
if err != nil {
|
|
t.Fatalf("readResourceGroupConfig(nil): %v", err)
|
|
}
|
|
if cfg.GroupBy != resourceGroupNone {
|
|
t.Fatalf("expected default groupBy=%q, got %q", resourceGroupNone, cfg.GroupBy)
|
|
}
|
|
|
|
cfg, err = readResourceGroupConfig(map[string]*plugin_pb.ConfigValue{
|
|
"resource_group_by": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: "bucket_namespace"}},
|
|
"max_tables_per_resource_group": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 2}},
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("readResourceGroupConfig(valid): %v", err)
|
|
}
|
|
if cfg.GroupBy != resourceGroupBucketNamespace {
|
|
t.Fatalf("expected bucket_namespace grouping, got %q", cfg.GroupBy)
|
|
}
|
|
if cfg.MaxTablesPerGroup != 2 {
|
|
t.Fatalf("expected max tables per group=2, got %d", cfg.MaxTablesPerGroup)
|
|
}
|
|
|
|
if _, err := readResourceGroupConfig(map[string]*plugin_pb.ConfigValue{
|
|
"resource_group_by": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: "invalid"}},
|
|
}); err == nil {
|
|
t.Fatal("expected invalid resource_group_by to fail")
|
|
}
|
|
|
|
if _, err := readResourceGroupConfig(map[string]*plugin_pb.ConfigValue{
|
|
"max_tables_per_resource_group": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 1}},
|
|
}); err == nil {
|
|
t.Fatal("expected group cap without grouping to fail")
|
|
}
|
|
}
|
|
|
|
func TestSelectTablesByResourceGroupRoundRobin(t *testing.T) {
|
|
tables := []tableInfo{
|
|
{BucketName: "a", Namespace: "ns1", TableName: "t1"},
|
|
{BucketName: "a", Namespace: "ns1", TableName: "t2"},
|
|
{BucketName: "b", Namespace: "ns2", TableName: "t3"},
|
|
{BucketName: "b", Namespace: "ns2", TableName: "t4"},
|
|
}
|
|
|
|
selected, hasMore := selectTablesByResourceGroup(tables, resourceGroupConfig{
|
|
GroupBy: resourceGroupBucket,
|
|
}, 3)
|
|
if !hasMore {
|
|
t.Fatal("expected hasMore when maxResults truncates the selection")
|
|
}
|
|
if len(selected) != 3 {
|
|
t.Fatalf("expected 3 selected tables, got %d", len(selected))
|
|
}
|
|
if selected[0].BucketName != "a" || selected[1].BucketName != "b" || selected[2].BucketName != "a" {
|
|
t.Fatalf("expected round-robin bucket order [a, b, a], got [%s, %s, %s]", selected[0].BucketName, selected[1].BucketName, selected[2].BucketName)
|
|
}
|
|
}
|
|
|
|
func TestSelectTablesByResourceGroupCap(t *testing.T) {
|
|
tables := []tableInfo{
|
|
{BucketName: "a", Namespace: "ns1", TableName: "t1"},
|
|
{BucketName: "a", Namespace: "ns1", TableName: "t2"},
|
|
{BucketName: "b", Namespace: "ns2", TableName: "t3"},
|
|
{BucketName: "b", Namespace: "ns2", TableName: "t4"},
|
|
}
|
|
|
|
selected, hasMore := selectTablesByResourceGroup(tables, resourceGroupConfig{
|
|
GroupBy: resourceGroupBucket,
|
|
MaxTablesPerGroup: 1,
|
|
}, 0)
|
|
if !hasMore {
|
|
t.Fatal("expected hasMore when per-group cap omits tables")
|
|
}
|
|
if len(selected) != 2 {
|
|
t.Fatalf("expected 2 selected tables, got %d", len(selected))
|
|
}
|
|
if selected[0].BucketName != "a" || selected[1].BucketName != "b" {
|
|
t.Fatalf("expected one table per bucket, got [%s, %s]", selected[0].BucketName, selected[1].BucketName)
|
|
}
|
|
}
|
|
|
|
func TestManifestRewritePathConsistency(t *testing.T) {
|
|
// Verify that WriteManifest returns a ManifestFile whose FilePath()
|
|
// matches the path we pass in. This ensures the pattern used in
|
|
// rewriteManifests (compute filename once, pass to both WriteManifest
|
|
// and saveFilerFile) produces consistent references.
|
|
schema := newTestSchema()
|
|
spec := *iceberg.UnpartitionedSpec
|
|
|
|
snapshotID := int64(42)
|
|
manifestFileName := fmt.Sprintf("merged-%d-%d.avro", snapshotID, int64(1700000000000))
|
|
manifestPath := "metadata/" + manifestFileName
|
|
|
|
// Create a minimal manifest entry to write
|
|
dfBuilder, err := iceberg.NewDataFileBuilder(
|
|
spec,
|
|
iceberg.EntryContentData,
|
|
"data/test.parquet",
|
|
iceberg.ParquetFile,
|
|
map[int]any{},
|
|
nil, nil,
|
|
1, // recordCount
|
|
1024, // fileSizeBytes
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("failed to build data file: %v", err)
|
|
}
|
|
entry := iceberg.NewManifestEntry(
|
|
iceberg.EntryStatusADDED,
|
|
&snapshotID,
|
|
nil, nil,
|
|
dfBuilder.Build(),
|
|
)
|
|
|
|
var buf bytes.Buffer
|
|
mf, err := iceberg.WriteManifest(
|
|
manifestPath,
|
|
&buf,
|
|
2, // version
|
|
spec,
|
|
schema,
|
|
snapshotID,
|
|
[]iceberg.ManifestEntry{entry},
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("WriteManifest failed: %v", err)
|
|
}
|
|
|
|
if mf.FilePath() != manifestPath {
|
|
t.Errorf("manifest FilePath() = %q, want %q", mf.FilePath(), manifestPath)
|
|
}
|
|
|
|
// Verify the filename we'd use for saveFilerFile matches
|
|
if path.Base(mf.FilePath()) != manifestFileName {
|
|
t.Errorf("manifest base name = %q, want %q", path.Base(mf.FilePath()), manifestFileName)
|
|
}
|
|
}
|
|
|
|
func TestManifestRewriteNestedPathConsistency(t *testing.T) {
|
|
// Verify that WriteManifest with nested paths preserves the full path
|
|
// and that loadFileByIcebergPath (via normalizeIcebergPath) would
|
|
// resolve them correctly.
|
|
schema := newTestSchema()
|
|
spec := *iceberg.UnpartitionedSpec
|
|
snapshotID := int64(42)
|
|
|
|
testCases := []struct {
|
|
name string
|
|
manifestPath string
|
|
}{
|
|
{"nested two levels", "metadata/a/b/merged-42-1700000000000.avro"},
|
|
{"nested one level", "metadata/subdir/manifest-42.avro"},
|
|
}
|
|
|
|
for _, tc := range testCases {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
dfBuilder, err := iceberg.NewDataFileBuilder(
|
|
spec,
|
|
iceberg.EntryContentData,
|
|
"data/test.parquet",
|
|
iceberg.ParquetFile,
|
|
map[int]any{},
|
|
nil, nil,
|
|
1, 1024,
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("failed to build data file: %v", err)
|
|
}
|
|
entry := iceberg.NewManifestEntry(
|
|
iceberg.EntryStatusADDED,
|
|
&snapshotID,
|
|
nil, nil,
|
|
dfBuilder.Build(),
|
|
)
|
|
|
|
var buf bytes.Buffer
|
|
mf, err := iceberg.WriteManifest(
|
|
tc.manifestPath, &buf, 2, spec, schema, snapshotID,
|
|
[]iceberg.ManifestEntry{entry},
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("WriteManifest failed: %v", err)
|
|
}
|
|
|
|
if mf.FilePath() != tc.manifestPath {
|
|
t.Errorf("FilePath() = %q, want %q", mf.FilePath(), tc.manifestPath)
|
|
}
|
|
|
|
// A relative path resolves under the table's own directory
|
|
want := "ns/table/" + tc.manifestPath
|
|
normalized, err := normalizeIcebergPath(tc.manifestPath, "bucket", "ns/table")
|
|
if err != nil {
|
|
t.Fatalf("normalizeIcebergPath(%q): %v", tc.manifestPath, err)
|
|
}
|
|
if normalized != want {
|
|
t.Errorf("normalizeIcebergPath(%q) = %q, want %q", tc.manifestPath, normalized, want)
|
|
}
|
|
|
|
// The S3 URL for the same file resolves to the same place
|
|
s3Path := "s3://bucket/ns/table/" + tc.manifestPath
|
|
normalized, err = normalizeIcebergPath(s3Path, "bucket", "ns/table")
|
|
if err != nil {
|
|
t.Fatalf("normalizeIcebergPath(%q): %v", s3Path, err)
|
|
}
|
|
if normalized != want {
|
|
t.Errorf("normalizeIcebergPath(%q) = %q, want %q", s3Path, normalized, want)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestNormalizeIcebergPath(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
icebergPath string
|
|
bucket string
|
|
dataPath string
|
|
expected string
|
|
wantErr bool
|
|
}{
|
|
{
|
|
name: "relative metadata path",
|
|
icebergPath: "metadata/snap-1.avro",
|
|
bucket: "mybucket", dataPath: "ns/table",
|
|
expected: "ns/table/metadata/snap-1.avro",
|
|
},
|
|
{
|
|
name: "relative data path",
|
|
icebergPath: "data/file.parquet",
|
|
bucket: "mybucket", dataPath: "ns/table",
|
|
expected: "ns/table/data/file.parquet",
|
|
},
|
|
{
|
|
name: "S3 URL",
|
|
icebergPath: "s3://mybucket/ns/table/metadata/snap-1.avro",
|
|
bucket: "mybucket", dataPath: "ns/table",
|
|
expected: "ns/table/metadata/snap-1.avro",
|
|
},
|
|
{
|
|
name: "absolute filer path",
|
|
icebergPath: "/buckets/mybucket/ns/table/data/file.parquet",
|
|
bucket: "mybucket", dataPath: "ns/table",
|
|
expected: "ns/table/data/file.parquet",
|
|
},
|
|
{
|
|
name: "nested data path",
|
|
icebergPath: "data/region=us/city=sf/file.parquet",
|
|
bucket: "mybucket", dataPath: "ns/table",
|
|
expected: "ns/table/data/region=us/city=sf/file.parquet",
|
|
},
|
|
{
|
|
name: "S3 URL nested",
|
|
icebergPath: "s3://mybucket/ns/table/data/region=us/file.parquet",
|
|
bucket: "mybucket", dataPath: "ns/table",
|
|
expected: "ns/table/data/region=us/file.parquet",
|
|
},
|
|
{
|
|
// Written by a client that created the table at its own location.
|
|
name: "S3 URL outside the catalog path",
|
|
icebergPath: "s3://mybucket/ns/table-9f1c/metadata/snap-1.avro",
|
|
bucket: "mybucket", dataPath: "ns/table-9f1c",
|
|
expected: "ns/table-9f1c/metadata/snap-1.avro",
|
|
},
|
|
{
|
|
// Same file, referenced while the worker believes the table sits at
|
|
// its catalog path: the URI still wins.
|
|
name: "S3 URL wins over a stale data path",
|
|
icebergPath: "s3://mybucket/ns/table-9f1c/metadata/snap-1.avro",
|
|
bucket: "mybucket", dataPath: "ns/table",
|
|
expected: "ns/table-9f1c/metadata/snap-1.avro",
|
|
},
|
|
{
|
|
name: "other bucket",
|
|
icebergPath: "s3://otherbucket/ns/table/metadata/snap-1.avro",
|
|
bucket: "mybucket", dataPath: "ns/table",
|
|
wantErr: true,
|
|
},
|
|
{
|
|
name: "traversal out of the bucket",
|
|
icebergPath: "../../etc/passwd",
|
|
bucket: "mybucket", dataPath: "ns/table",
|
|
wantErr: true,
|
|
},
|
|
}
|
|
|
|
for _, tc := range tests {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
result, err := normalizeIcebergPath(tc.icebergPath, tc.bucket, tc.dataPath)
|
|
if tc.wantErr {
|
|
if err == nil {
|
|
t.Fatalf("normalizeIcebergPath(%q, %q, %q) = %q, want error",
|
|
tc.icebergPath, tc.bucket, tc.dataPath, result)
|
|
}
|
|
return
|
|
}
|
|
if err != nil {
|
|
t.Fatalf("normalizeIcebergPath(%q, %q, %q): %v", tc.icebergPath, tc.bucket, tc.dataPath, err)
|
|
}
|
|
if result != tc.expected {
|
|
t.Errorf("normalizeIcebergPath(%q, %q, %q) = %q, want %q",
|
|
tc.icebergPath, tc.bucket, tc.dataPath, result, tc.expected)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestTableDataPath(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
metadataLocation string
|
|
expected string
|
|
}{
|
|
{
|
|
name: "catalog path",
|
|
metadataLocation: "s3://mybucket/ns/table/metadata/v1.metadata.json",
|
|
expected: "ns/table",
|
|
},
|
|
{
|
|
name: "location outside the catalog path",
|
|
metadataLocation: "s3://mybucket/ns/table-9f1c/metadata/v3-171.metadata.json",
|
|
expected: "ns/table-9f1c",
|
|
},
|
|
{
|
|
name: "relative location",
|
|
metadataLocation: "metadata/v1.metadata.json",
|
|
expected: "ns/table",
|
|
},
|
|
{
|
|
name: "missing location",
|
|
metadataLocation: "",
|
|
expected: "ns/table",
|
|
},
|
|
{
|
|
name: "location in another bucket",
|
|
metadataLocation: "s3://otherbucket/ns/table/metadata/v1.metadata.json",
|
|
expected: "ns/table",
|
|
},
|
|
}
|
|
|
|
for _, tc := range tests {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
if got := tableDataPath("mybucket", "ns/table", tc.metadataLocation); got != tc.expected {
|
|
t.Errorf("tableDataPath(%q) = %q, want %q", tc.metadataLocation, got, tc.expected)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestPartitionKey(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
partition map[int]any
|
|
expected string
|
|
}{
|
|
{"empty partition", map[int]any{}, "__unpartitioned__"},
|
|
{"nil partition", nil, "__unpartitioned__"},
|
|
{"single field", map[int]any{1: "us-east"}, "1=\"us-east\""},
|
|
{"multiple fields sorted", map[int]any{3: "2024", 1: "us-east"}, "1=\"us-east\"\x003=\"2024\""},
|
|
}
|
|
|
|
for _, tc := range tests {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
result := partitionKey(tc.partition)
|
|
if result != tc.expected {
|
|
t.Errorf("partitionKey(%v) = %q, want %q", tc.partition, result, tc.expected)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestBuildCompactionBins(t *testing.T) {
|
|
targetSize := int64(256 * 1024 * 1024) // 256MB
|
|
minFiles := 3
|
|
|
|
// Create test entries: small files in same partition
|
|
entries := makeTestEntries(t, []testEntrySpec{
|
|
{path: "data/f1.parquet", size: 1024, partition: map[int]any{}},
|
|
{path: "data/f2.parquet", size: 2048, partition: map[int]any{}},
|
|
{path: "data/f3.parquet", size: 4096, partition: map[int]any{}},
|
|
})
|
|
|
|
bins := buildCompactionBins(entries, targetSize, minFiles)
|
|
if len(bins) != 1 {
|
|
t.Fatalf("expected 1 bin, got %d", len(bins))
|
|
}
|
|
if len(bins[0].Entries) != 3 {
|
|
t.Errorf("expected 3 entries in bin, got %d", len(bins[0].Entries))
|
|
}
|
|
}
|
|
|
|
func TestBuildCompactionBinsFiltersLargeFiles(t *testing.T) {
|
|
targetSize := int64(4000)
|
|
minFiles := 2
|
|
|
|
entries := makeTestEntries(t, []testEntrySpec{
|
|
{path: "data/small1.parquet", size: 1024, partition: map[int]any{}},
|
|
{path: "data/small2.parquet", size: 2048, partition: map[int]any{}},
|
|
{path: "data/large.parquet", size: 5000, partition: map[int]any{}},
|
|
})
|
|
|
|
bins := buildCompactionBins(entries, targetSize, minFiles)
|
|
if len(bins) != 1 {
|
|
t.Fatalf("expected 1 bin, got %d", len(bins))
|
|
}
|
|
if len(bins[0].Entries) != 2 {
|
|
t.Errorf("expected 2 entries (large excluded), got %d", len(bins[0].Entries))
|
|
}
|
|
}
|
|
|
|
func TestBuildCompactionBinsMinFilesThreshold(t *testing.T) {
|
|
targetSize := int64(256 * 1024 * 1024)
|
|
minFiles := 5
|
|
|
|
entries := makeTestEntries(t, []testEntrySpec{
|
|
{path: "data/f1.parquet", size: 1024, partition: map[int]any{}},
|
|
{path: "data/f2.parquet", size: 2048, partition: map[int]any{}},
|
|
})
|
|
|
|
bins := buildCompactionBins(entries, targetSize, minFiles)
|
|
if len(bins) != 0 {
|
|
t.Errorf("expected 0 bins (below min threshold), got %d", len(bins))
|
|
}
|
|
}
|
|
|
|
func TestBuildCompactionBinsMultiplePartitions(t *testing.T) {
|
|
targetSize := int64(256 * 1024 * 1024)
|
|
minFiles := 2
|
|
|
|
partA := map[int]any{1: "us-east"}
|
|
partB := map[int]any{1: "eu-west"}
|
|
partitionSpec := iceberg.NewPartitionSpec(iceberg.PartitionField{
|
|
SourceIDs: []int{1},
|
|
FieldID: 1000,
|
|
Name: "region",
|
|
Transform: iceberg.IdentityTransform{},
|
|
})
|
|
|
|
entries := makeTestEntries(t, []testEntrySpec{
|
|
{path: "data/a1.parquet", size: 1024, partition: partA, partitionSpec: &partitionSpec},
|
|
{path: "data/a2.parquet", size: 2048, partition: partA, partitionSpec: &partitionSpec},
|
|
{path: "data/b1.parquet", size: 1024, partition: partB, partitionSpec: &partitionSpec},
|
|
{path: "data/b2.parquet", size: 2048, partition: partB, partitionSpec: &partitionSpec},
|
|
{path: "data/b3.parquet", size: 4096, partition: partB, partitionSpec: &partitionSpec},
|
|
})
|
|
|
|
bins := buildCompactionBins(entries, targetSize, minFiles)
|
|
if len(bins) != 2 {
|
|
t.Fatalf("expected 2 bins (one per partition), got %d", len(bins))
|
|
}
|
|
}
|
|
|
|
func TestSplitOversizedBinRespectsTargetSize(t *testing.T) {
|
|
targetSize := int64(100)
|
|
minFiles := 2
|
|
|
|
entries := makeTestEntries(t, []testEntrySpec{
|
|
{path: "data/f1.parquet", size: 80, partition: map[int]any{}},
|
|
{path: "data/f2.parquet", size: 80, partition: map[int]any{}},
|
|
{path: "data/f3.parquet", size: 10, partition: map[int]any{}},
|
|
{path: "data/f4.parquet", size: 10, partition: map[int]any{}},
|
|
})
|
|
|
|
bins := splitOversizedBin(compactionBin{
|
|
PartitionKey: "__unpartitioned__",
|
|
Partition: map[int]any{},
|
|
Entries: entries,
|
|
TotalSize: 180,
|
|
}, targetSize, minFiles)
|
|
|
|
if len(bins) == 0 {
|
|
t.Fatal("expected at least one valid bin")
|
|
}
|
|
for i, bin := range bins {
|
|
if bin.TotalSize > targetSize {
|
|
t.Fatalf("bin %d exceeds target size: got %d want <= %d", i, bin.TotalSize, targetSize)
|
|
}
|
|
if len(bin.Entries) < minFiles {
|
|
t.Fatalf("bin %d does not meet minFiles: got %d want >= %d", i, len(bin.Entries), minFiles)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestSplitOversizedBinDropsImpossibleRunts(t *testing.T) {
|
|
targetSize := int64(100)
|
|
minFiles := 2
|
|
|
|
entries := makeTestEntries(t, []testEntrySpec{
|
|
{path: "data/f1.parquet", size: 60, partition: map[int]any{}},
|
|
{path: "data/f2.parquet", size: 60, partition: map[int]any{}},
|
|
{path: "data/f3.parquet", size: 60, partition: map[int]any{}},
|
|
})
|
|
|
|
bins := splitOversizedBin(compactionBin{
|
|
PartitionKey: "__unpartitioned__",
|
|
Partition: map[int]any{},
|
|
Entries: entries,
|
|
TotalSize: 180,
|
|
}, targetSize, minFiles)
|
|
|
|
if len(bins) != 0 {
|
|
t.Fatalf("expected no valid bins, got %d", len(bins))
|
|
}
|
|
}
|
|
|
|
type testEntrySpec struct {
|
|
path string
|
|
size int64
|
|
partition map[int]any
|
|
partitionSpec *iceberg.PartitionSpec
|
|
specID int32 // partition spec ID; 0 uses UnpartitionedSpec
|
|
}
|
|
|
|
func buildTestDataFile(t *testing.T, spec testEntrySpec) iceberg.DataFile {
|
|
t.Helper()
|
|
|
|
partitionSpec := iceberg.UnpartitionedSpec
|
|
if spec.partitionSpec != nil {
|
|
partitionSpec = spec.partitionSpec
|
|
} else if len(spec.partition) > 0 {
|
|
t.Fatalf("partition spec is required for partitioned test entry %s", spec.path)
|
|
}
|
|
dfBuilder, err := iceberg.NewDataFileBuilder(
|
|
*partitionSpec,
|
|
iceberg.EntryContentData,
|
|
spec.path,
|
|
iceberg.ParquetFile,
|
|
spec.partition,
|
|
nil, nil,
|
|
1, // recordCount (must be > 0)
|
|
spec.size,
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("failed to build data file %s: %v", spec.path, err)
|
|
}
|
|
return dfBuilder.Build()
|
|
}
|
|
|
|
func makeManifestEntriesWithSnapshot(
|
|
t *testing.T,
|
|
specs []testEntrySpec,
|
|
snapshotID int64,
|
|
status iceberg.ManifestEntryStatus,
|
|
) []iceberg.ManifestEntry {
|
|
t.Helper()
|
|
|
|
entries := make([]iceberg.ManifestEntry, 0, len(specs))
|
|
for _, spec := range specs {
|
|
entries = append(entries, iceberg.NewManifestEntry(
|
|
status,
|
|
&snapshotID,
|
|
nil, nil,
|
|
buildTestDataFile(t, spec),
|
|
))
|
|
}
|
|
return entries
|
|
}
|
|
|
|
// makeTestEntries creates manifest entries using the default unpartitioned
|
|
// spec. For multi-spec testing, use makeTestEntriesWithSpec instead.
|
|
func makeTestEntries(t *testing.T, specs []testEntrySpec) []iceberg.ManifestEntry {
|
|
t.Helper()
|
|
return makeManifestEntriesWithSnapshot(t, specs, 1, iceberg.EntryStatusADDED)
|
|
}
|
|
|
|
// makeTestEntriesWithSpec creates manifest entries using specific partition specs.
|
|
// Each spec in the specs slice can specify a specID; the entry is built using
|
|
// a PartitionSpec with that ID.
|
|
func makeTestEntriesWithSpec(t *testing.T, specs []testEntrySpec, partSpecs map[int32]iceberg.PartitionSpec) []iceberg.ManifestEntry {
|
|
t.Helper()
|
|
entries := make([]iceberg.ManifestEntry, 0, len(specs))
|
|
for _, s := range specs {
|
|
ps, ok := partSpecs[s.specID]
|
|
if !ok {
|
|
t.Fatalf("spec ID %d not found in partSpecs map", s.specID)
|
|
}
|
|
dfBuilder, err := iceberg.NewDataFileBuilder(
|
|
ps,
|
|
iceberg.EntryContentData,
|
|
s.path,
|
|
iceberg.ParquetFile,
|
|
s.partition,
|
|
nil, nil,
|
|
1,
|
|
s.size,
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("failed to build data file %s: %v", s.path, err)
|
|
}
|
|
snapID := int64(1)
|
|
entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfBuilder.Build())
|
|
entries = append(entries, entry)
|
|
}
|
|
return entries
|
|
}
|
|
|
|
func TestBuildCompactionBinsMultipleSpecs(t *testing.T) {
|
|
targetSize := int64(256 * 1024 * 1024)
|
|
minFiles := 2
|
|
|
|
// Two partition specs with different IDs
|
|
spec0 := iceberg.NewPartitionSpecID(0)
|
|
spec1 := iceberg.NewPartitionSpecID(1)
|
|
partSpecs := map[int32]iceberg.PartitionSpec{
|
|
0: spec0,
|
|
1: spec1,
|
|
}
|
|
|
|
// Entries from two different specs with same partition values should go
|
|
// to separate bins.
|
|
entries := makeTestEntriesWithSpec(t, []testEntrySpec{
|
|
{path: "data/s0-f1.parquet", size: 1024, partition: map[int]any{}, specID: 0},
|
|
{path: "data/s0-f2.parquet", size: 2048, partition: map[int]any{}, specID: 0},
|
|
{path: "data/s1-f1.parquet", size: 1024, partition: map[int]any{}, specID: 1},
|
|
{path: "data/s1-f2.parquet", size: 2048, partition: map[int]any{}, specID: 1},
|
|
}, partSpecs)
|
|
|
|
bins := buildCompactionBins(entries, targetSize, minFiles)
|
|
if len(bins) != 2 {
|
|
t.Fatalf("expected 2 bins (one per spec), got %d", len(bins))
|
|
}
|
|
|
|
// Verify each bin has entries from only one spec
|
|
specsSeen := make(map[int32]bool)
|
|
for _, bin := range bins {
|
|
specsSeen[bin.SpecID] = true
|
|
for _, entry := range bin.Entries {
|
|
if entry.DataFile().SpecID() != bin.SpecID {
|
|
t.Errorf("bin specID=%d contains entry with specID=%d", bin.SpecID, entry.DataFile().SpecID())
|
|
}
|
|
}
|
|
}
|
|
if !specsSeen[0] || !specsSeen[1] {
|
|
t.Errorf("expected bins for spec 0 and 1, got specs %v", specsSeen)
|
|
}
|
|
}
|
|
|
|
func TestBuildCompactionBinsSingleSpec(t *testing.T) {
|
|
// Verify existing behavior: all entries with spec 0 still group correctly.
|
|
targetSize := int64(256 * 1024 * 1024)
|
|
minFiles := 2
|
|
|
|
spec0 := iceberg.NewPartitionSpecID(0)
|
|
partSpecs := map[int32]iceberg.PartitionSpec{0: spec0}
|
|
|
|
entries := makeTestEntriesWithSpec(t, []testEntrySpec{
|
|
{path: "data/f1.parquet", size: 1024, partition: map[int]any{}, specID: 0},
|
|
{path: "data/f2.parquet", size: 2048, partition: map[int]any{}, specID: 0},
|
|
{path: "data/f3.parquet", size: 4096, partition: map[int]any{}, specID: 0},
|
|
}, partSpecs)
|
|
|
|
bins := buildCompactionBins(entries, targetSize, minFiles)
|
|
if len(bins) != 1 {
|
|
t.Fatalf("expected 1 bin, got %d", len(bins))
|
|
}
|
|
if len(bins[0].Entries) != 3 {
|
|
t.Errorf("expected 3 entries in bin, got %d", len(bins[0].Entries))
|
|
}
|
|
if bins[0].SpecID != 0 {
|
|
t.Errorf("expected specID=0, got %d", bins[0].SpecID)
|
|
}
|
|
}
|
|
|
|
func TestParseConfigApplyDeletes(t *testing.T) {
|
|
// Default: true
|
|
config := ParseConfig(nil)
|
|
if !config.ApplyDeletes {
|
|
t.Error("expected ApplyDeletes=true by default")
|
|
}
|
|
|
|
// Explicit false
|
|
config = ParseConfig(map[string]*plugin_pb.ConfigValue{
|
|
"apply_deletes": {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: false}},
|
|
})
|
|
if config.ApplyDeletes {
|
|
t.Error("expected ApplyDeletes=false when explicitly set")
|
|
}
|
|
|
|
// String "false"
|
|
config = ParseConfig(map[string]*plugin_pb.ConfigValue{
|
|
"apply_deletes": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: "false"}},
|
|
})
|
|
if config.ApplyDeletes {
|
|
t.Error("expected ApplyDeletes=false when set via string 'false'")
|
|
}
|
|
}
|
|
|
|
func TestNormalizeDetectionConfigUsesSharedDefaults(t *testing.T) {
|
|
config := normalizeDetectionConfig(Config{})
|
|
|
|
if config.TargetFileSizeBytes != defaultTargetFileSizeMB*1024*1024 {
|
|
t.Fatalf("expected TargetFileSizeBytes default, got %d", config.TargetFileSizeBytes)
|
|
}
|
|
if config.DeleteTargetFileSizeBytes != defaultDeleteTargetFileSizeMB*1024*1024 {
|
|
t.Fatalf("expected DeleteTargetFileSizeBytes default, got %d", config.DeleteTargetFileSizeBytes)
|
|
}
|
|
if config.DeleteMinInputFiles != defaultDeleteMinInputFiles {
|
|
t.Fatalf("expected DeleteMinInputFiles default, got %d", config.DeleteMinInputFiles)
|
|
}
|
|
if config.DeleteMaxFileGroupSizeBytes != defaultDeleteMaxGroupSizeMB*1024*1024 {
|
|
t.Fatalf("expected DeleteMaxFileGroupSizeBytes default, got %d", config.DeleteMaxFileGroupSizeBytes)
|
|
}
|
|
if config.DeleteMaxOutputFiles != defaultDeleteMaxOutputFiles {
|
|
t.Fatalf("expected DeleteMaxOutputFiles default, got %d", config.DeleteMaxOutputFiles)
|
|
}
|
|
if config.OrphanOlderThanHours != defaultOrphanOlderThanHours {
|
|
t.Fatalf("expected OrphanOlderThanHours default, got %d", config.OrphanOlderThanHours)
|
|
}
|
|
if config.SnapshotRetentionHours != defaultSnapshotRetentionHours {
|
|
t.Fatalf("expected SnapshotRetentionHours default, got %d", config.SnapshotRetentionHours)
|
|
}
|
|
if config.MaxSnapshotsToKeep != defaultMaxSnapshotsToKeep {
|
|
t.Fatalf("expected MaxSnapshotsToKeep default, got %d", config.MaxSnapshotsToKeep)
|
|
}
|
|
}
|
|
|
|
func TestParseConfigRewriteStrategy(t *testing.T) {
|
|
config := ParseConfig(map[string]*plugin_pb.ConfigValue{
|
|
"rewrite_strategy": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: "sort"}},
|
|
"sort_max_input_mb": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 64}},
|
|
})
|
|
if config.RewriteStrategy != "sort" {
|
|
t.Fatalf("expected sort rewrite strategy, got %q", config.RewriteStrategy)
|
|
}
|
|
if config.SortMaxInputBytes != 64*1024*1024 {
|
|
t.Fatalf("expected SortMaxInputBytes=64MB, got %d", config.SortMaxInputBytes)
|
|
}
|
|
|
|
config = ParseConfig(map[string]*plugin_pb.ConfigValue{
|
|
"rewrite_strategy": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: "invalid"}},
|
|
"sort_max_input_mb": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: -1}},
|
|
})
|
|
if config.RewriteStrategy != defaultRewriteStrategy {
|
|
t.Fatalf("expected invalid rewrite strategy to fall back to %q, got %q", defaultRewriteStrategy, config.RewriteStrategy)
|
|
}
|
|
if config.SortMaxInputBytes != 0 {
|
|
t.Fatalf("expected negative sort cap to clamp to 0, got %d", config.SortMaxInputBytes)
|
|
}
|
|
|
|
maxMB := int64(^uint64(0)>>1) / bytesPerMB
|
|
config = ParseConfig(map[string]*plugin_pb.ConfigValue{
|
|
"sort_max_input_mb": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: fmt.Sprintf("%d", maxMB+1)}},
|
|
})
|
|
if config.SortMaxInputBytes != maxMB*bytesPerMB {
|
|
t.Fatalf("expected oversized sort cap to clamp to %d bytes, got %d", maxMB*bytesPerMB, config.SortMaxInputBytes)
|
|
}
|
|
}
|
|
|
|
func TestCollectPositionDeletes(t *testing.T) {
|
|
fs, client := startFakeFiler(t)
|
|
|
|
// Create a position delete Parquet file with file_path + pos columns
|
|
type posDeleteRow struct {
|
|
FilePath string `parquet:"file_path"`
|
|
Pos int64 `parquet:"pos"`
|
|
}
|
|
var buf bytes.Buffer
|
|
writer := parquet.NewWriter(&buf, parquet.SchemaOf(new(posDeleteRow)))
|
|
rows := []posDeleteRow{
|
|
{"data/file1.parquet", 0},
|
|
{"data/file1.parquet", 5},
|
|
{"data/file1.parquet", 10},
|
|
{"data/file2.parquet", 3},
|
|
}
|
|
for _, r := range rows {
|
|
if err := writer.Write(&r); err != nil {
|
|
t.Fatalf("write pos delete row: %v", err)
|
|
}
|
|
}
|
|
if err := writer.Close(); err != nil {
|
|
t.Fatalf("close pos delete writer: %v", err)
|
|
}
|
|
|
|
// Store in fake filer
|
|
dataDir := "/buckets/test-bucket/ns/tbl/data"
|
|
fs.putEntry(dataDir, "pos-delete-1.parquet", &filer_pb.Entry{
|
|
Name: "pos-delete-1.parquet",
|
|
Content: buf.Bytes(),
|
|
})
|
|
|
|
// Create a manifest entry for the position delete file
|
|
spec := *iceberg.UnpartitionedSpec
|
|
dfBuilder, err := iceberg.NewDataFileBuilder(
|
|
spec,
|
|
iceberg.EntryContentPosDeletes,
|
|
"data/pos-delete-1.parquet",
|
|
iceberg.ParquetFile,
|
|
map[int]any{},
|
|
nil, nil,
|
|
int64(len(rows)),
|
|
int64(buf.Len()),
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("build pos delete data file: %v", err)
|
|
}
|
|
snapID := int64(1)
|
|
entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfBuilder.Build())
|
|
|
|
result, err := collectPositionDeletes(context.Background(), client, "test-bucket", "ns/tbl", []iceberg.ManifestEntry{entry})
|
|
if err != nil {
|
|
t.Fatalf("collectPositionDeletes: %v", err)
|
|
}
|
|
|
|
// Deleted files are keyed by their bucket-relative path
|
|
if len(result["ns/tbl/data/file1.parquet"]) != 3 {
|
|
t.Errorf("expected 3 positions for file1, got %d", len(result["ns/tbl/data/file1.parquet"]))
|
|
}
|
|
if len(result["ns/tbl/data/file2.parquet"]) != 1 {
|
|
t.Errorf("expected 1 position for file2, got %d", len(result["ns/tbl/data/file2.parquet"]))
|
|
}
|
|
|
|
// Verify sorted
|
|
positions := result["ns/tbl/data/file1.parquet"]
|
|
for i := 1; i < len(positions); i++ {
|
|
if positions[i] <= positions[i-1] {
|
|
t.Errorf("positions not sorted: %v", positions)
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestCollectEqualityDeletes(t *testing.T) {
|
|
fs, client := startFakeFiler(t)
|
|
|
|
// Create an equality delete Parquet file with id + name columns
|
|
type eqDeleteRow struct {
|
|
ID int64 `parquet:"id"`
|
|
Name string `parquet:"name"`
|
|
}
|
|
var buf bytes.Buffer
|
|
writer := parquet.NewWriter(&buf, parquet.SchemaOf(new(eqDeleteRow)))
|
|
deleteRows := []eqDeleteRow{
|
|
{1, "alice"},
|
|
{2, "bob"},
|
|
{3, "charlie"},
|
|
}
|
|
for _, r := range deleteRows {
|
|
if err := writer.Write(&r); err != nil {
|
|
t.Fatalf("write eq delete row: %v", err)
|
|
}
|
|
}
|
|
if err := writer.Close(); err != nil {
|
|
t.Fatalf("close eq delete writer: %v", err)
|
|
}
|
|
|
|
dataDir := "/buckets/test-bucket/ns/tbl/data"
|
|
fs.putEntry(dataDir, "eq-delete-1.parquet", &filer_pb.Entry{
|
|
Name: "eq-delete-1.parquet",
|
|
Content: buf.Bytes(),
|
|
})
|
|
|
|
// Create manifest entry with equality field IDs
|
|
schema := newTestSchema() // has field 1=id, 2=name
|
|
spec := *iceberg.UnpartitionedSpec
|
|
dfBuilder, err := iceberg.NewDataFileBuilder(
|
|
spec,
|
|
iceberg.EntryContentEqDeletes,
|
|
"data/eq-delete-1.parquet",
|
|
iceberg.ParquetFile,
|
|
map[int]any{},
|
|
nil, nil,
|
|
int64(len(deleteRows)),
|
|
int64(buf.Len()),
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("build eq delete data file: %v", err)
|
|
}
|
|
dfBuilder.EqualityFieldIDs([]int{1, 2}) // id + name
|
|
snapID := int64(1)
|
|
entry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfBuilder.Build())
|
|
|
|
groups, err := collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl", []iceberg.ManifestEntry{entry}, schema)
|
|
if err != nil {
|
|
t.Fatalf("collectEqualityDeletes: %v", err)
|
|
}
|
|
|
|
if len(groups) != 1 {
|
|
t.Fatalf("expected 1 equality delete group, got %d", len(groups))
|
|
}
|
|
if len(groups[0].Keys) != 3 {
|
|
t.Errorf("expected 3 equality delete keys, got %d", len(groups[0].Keys))
|
|
}
|
|
if len(groups[0].FieldIDs) != 2 {
|
|
t.Errorf("expected 2 field IDs, got %d", len(groups[0].FieldIDs))
|
|
}
|
|
}
|
|
|
|
func TestMergeParquetFilesWithPositionDeletes(t *testing.T) {
|
|
fs, client := startFakeFiler(t)
|
|
|
|
// Create two data files with known rows
|
|
type dataRow struct {
|
|
ID int64 `parquet:"id"`
|
|
Name string `parquet:"name"`
|
|
}
|
|
|
|
writeDataFile := func(name string, rows []dataRow) {
|
|
var buf bytes.Buffer
|
|
w := parquet.NewWriter(&buf, parquet.SchemaOf(new(dataRow)))
|
|
for _, r := range rows {
|
|
if err := w.Write(&r); err != nil {
|
|
t.Fatalf("write row: %v", err)
|
|
}
|
|
}
|
|
if err := w.Close(); err != nil {
|
|
t.Fatalf("close: %v", err)
|
|
}
|
|
dataDir := "/buckets/test-bucket/ns/tbl/data"
|
|
fs.putEntry(dataDir, name, &filer_pb.Entry{
|
|
Name: name,
|
|
Content: buf.Bytes(),
|
|
})
|
|
}
|
|
|
|
writeDataFile("file1.parquet", []dataRow{
|
|
{1, "alice"}, {2, "bob"}, {3, "charlie"}, {4, "dave"}, {5, "eve"},
|
|
})
|
|
writeDataFile("file2.parquet", []dataRow{
|
|
{6, "frank"}, {7, "grace"},
|
|
})
|
|
|
|
spec := *iceberg.UnpartitionedSpec
|
|
makeEntry := func(path string, size int64) iceberg.ManifestEntry {
|
|
dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData, path, iceberg.ParquetFile, map[int]any{}, nil, nil, 1, size)
|
|
if err != nil {
|
|
t.Fatalf("build: %v", err)
|
|
}
|
|
snapID := int64(1)
|
|
return iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfb.Build())
|
|
}
|
|
|
|
entries := []iceberg.ManifestEntry{
|
|
makeEntry("data/file1.parquet", 1024),
|
|
makeEntry("data/file2.parquet", 512),
|
|
}
|
|
|
|
// Delete rows 1 (bob) and 3 (dave) from file1
|
|
posDeletes := map[string][]int64{
|
|
"ns/tbl/data/file1.parquet": {1, 3},
|
|
}
|
|
|
|
merged, count, err := mergeParquetFiles(
|
|
context.Background(), client, "test-bucket", "ns/tbl",
|
|
entries, posDeletes, nil, nil,
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("mergeParquetFiles: %v", err)
|
|
}
|
|
|
|
// Original: 5 + 2 = 7 rows, deleted 2 = 5 rows
|
|
if count != 5 {
|
|
t.Errorf("expected 5 rows after position deletes, got %d", count)
|
|
}
|
|
|
|
// Verify merged output is valid Parquet
|
|
reader := parquet.NewReader(bytes.NewReader(merged))
|
|
defer reader.Close()
|
|
var outputRows []dataRow
|
|
for {
|
|
var r dataRow
|
|
err := reader.Read(&r)
|
|
if err == io.EOF {
|
|
break
|
|
}
|
|
if err != nil {
|
|
t.Fatalf("read merged row: %v", err)
|
|
}
|
|
outputRows = append(outputRows, r)
|
|
}
|
|
|
|
if len(outputRows) != 5 {
|
|
t.Fatalf("expected 5 output rows, got %d", len(outputRows))
|
|
}
|
|
|
|
// Verify bob (id=2) and dave (id=4) are NOT in output
|
|
for _, r := range outputRows {
|
|
if r.ID == 2 || r.ID == 4 {
|
|
t.Errorf("row with id=%d should have been deleted", r.ID)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestMergeParquetFilesWithEqualityDeletes(t *testing.T) {
|
|
fs, client := startFakeFiler(t)
|
|
|
|
type dataRow struct {
|
|
ID int64 `parquet:"id"`
|
|
Name string `parquet:"name"`
|
|
}
|
|
|
|
var buf bytes.Buffer
|
|
w := parquet.NewWriter(&buf, parquet.SchemaOf(new(dataRow)))
|
|
dataRows := []dataRow{
|
|
{1, "alice"}, {2, "bob"}, {3, "charlie"}, {4, "dave"},
|
|
}
|
|
for _, r := range dataRows {
|
|
if err := w.Write(&r); err != nil {
|
|
t.Fatalf("write: %v", err)
|
|
}
|
|
}
|
|
if err := w.Close(); err != nil {
|
|
t.Fatalf("close: %v", err)
|
|
}
|
|
|
|
dataDir := "/buckets/test-bucket/ns/tbl/data"
|
|
fs.putEntry(dataDir, "data1.parquet", &filer_pb.Entry{
|
|
Name: "data1.parquet",
|
|
Content: buf.Bytes(),
|
|
})
|
|
|
|
spec := *iceberg.UnpartitionedSpec
|
|
dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData, "data/data1.parquet", iceberg.ParquetFile, map[int]any{}, nil, nil, 4, int64(buf.Len()))
|
|
if err != nil {
|
|
t.Fatalf("build: %v", err)
|
|
}
|
|
snapID := int64(1)
|
|
entries := []iceberg.ManifestEntry{
|
|
iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfb.Build()),
|
|
}
|
|
|
|
// Build equality deletes via the collect pipeline so keys match the
|
|
// encoding used by mergeParquetFiles.
|
|
schema := newTestSchema() // field 1=id, field 2=name
|
|
|
|
type eqDeleteRow struct {
|
|
ID int64 `parquet:"id"`
|
|
Name string `parquet:"name"`
|
|
}
|
|
var eqBuf bytes.Buffer
|
|
eqWriter := parquet.NewWriter(&eqBuf, parquet.SchemaOf(new(eqDeleteRow)))
|
|
for _, r := range []eqDeleteRow{{0, "bob"}, {0, "dave"}} {
|
|
if err := eqWriter.Write(&r); err != nil {
|
|
t.Fatalf("write eq delete: %v", err)
|
|
}
|
|
}
|
|
if err := eqWriter.Close(); err != nil {
|
|
t.Fatalf("close eq writer: %v", err)
|
|
}
|
|
eqDataDir := "/buckets/test-bucket/ns/tbl/data"
|
|
fs.putEntry(eqDataDir, "eq-del.parquet", &filer_pb.Entry{
|
|
Name: "eq-del.parquet", Content: eqBuf.Bytes(),
|
|
})
|
|
|
|
eqDfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentEqDeletes,
|
|
"data/eq-del.parquet", iceberg.ParquetFile, map[int]any{}, nil, nil, 2, int64(eqBuf.Len()))
|
|
if err != nil {
|
|
t.Fatalf("build eq delete: %v", err)
|
|
}
|
|
eqDfb.EqualityFieldIDs([]int{2}) // field 2 = name
|
|
eqEntry := iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, eqDfb.Build())
|
|
|
|
eqGroups, err := collectEqualityDeletes(context.Background(), client, "test-bucket", "ns/tbl",
|
|
[]iceberg.ManifestEntry{eqEntry}, schema)
|
|
if err != nil {
|
|
t.Fatalf("collectEqualityDeletes: %v", err)
|
|
}
|
|
|
|
merged, count, err := mergeParquetFiles(
|
|
context.Background(), client, "test-bucket", "ns/tbl",
|
|
entries, nil, eqGroups, schema,
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("mergeParquetFiles: %v", err)
|
|
}
|
|
|
|
if count != 2 {
|
|
t.Errorf("expected 2 rows after equality deletes, got %d", count)
|
|
}
|
|
|
|
reader := parquet.NewReader(bytes.NewReader(merged))
|
|
defer reader.Close()
|
|
var outputRows []dataRow
|
|
for {
|
|
var r dataRow
|
|
err := reader.Read(&r)
|
|
if err == io.EOF {
|
|
break
|
|
}
|
|
if err != nil {
|
|
t.Fatalf("read: %v", err)
|
|
}
|
|
outputRows = append(outputRows, r)
|
|
}
|
|
|
|
for _, r := range outputRows {
|
|
if r.Name == "bob" || r.Name == "dave" {
|
|
t.Errorf("row %q should have been deleted", r.Name)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestDetectNilRequest(t *testing.T) {
|
|
handler := NewHandler(nil)
|
|
err := handler.Detect(nil, nil, nil)
|
|
if err == nil {
|
|
t.Error("expected error for nil request")
|
|
}
|
|
}
|
|
|
|
func TestExecuteNilRequest(t *testing.T) {
|
|
handler := NewHandler(nil)
|
|
err := handler.Execute(nil, nil, nil)
|
|
if err == nil {
|
|
t.Error("expected error for nil request")
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Test helpers
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// buildTestMetadata creates a minimal Iceberg metadata for testing.
|
|
// When snapshots is nil or empty, the metadata has no snapshots.
|
|
func buildTestMetadata(t *testing.T, snapshots []table.Snapshot) table.Metadata {
|
|
t.Helper()
|
|
|
|
schema := newTestSchema()
|
|
meta, err := table.NewMetadata(schema, iceberg.UnpartitionedSpec, table.UnsortedSortOrder, "s3://test-bucket/test-table", nil)
|
|
if err != nil {
|
|
t.Fatalf("failed to create test metadata: %v", err)
|
|
}
|
|
|
|
if len(snapshots) == 0 {
|
|
return meta
|
|
}
|
|
|
|
builder, err := table.MetadataBuilderFromBase(meta, "s3://test-bucket/test-table")
|
|
if err != nil {
|
|
t.Fatalf("failed to create metadata builder: %v", err)
|
|
}
|
|
|
|
var lastSnapID int64
|
|
for _, snap := range snapshots {
|
|
s := snap // copy
|
|
if err := builder.AddSnapshot(&s); err != nil {
|
|
t.Fatalf("failed to add snapshot %d: %v", snap.SnapshotID, err)
|
|
}
|
|
lastSnapID = snap.SnapshotID
|
|
}
|
|
|
|
if err := builder.SetSnapshotRef(table.MainBranch, lastSnapID, table.BranchRef); err != nil {
|
|
t.Fatalf("failed to set snapshot ref: %v", err)
|
|
}
|
|
|
|
result, err := builder.Build()
|
|
if err != nil {
|
|
t.Fatalf("failed to build metadata: %v", err)
|
|
}
|
|
return result
|
|
}
|
|
|
|
func newTestSchema() *iceberg.Schema {
|
|
return iceberg.NewSchema(0,
|
|
iceberg.NestedField{ID: 1, Type: iceberg.PrimitiveTypes.Int64, Name: "id", Required: true},
|
|
iceberg.NestedField{ID: 2, Type: iceberg.PrimitiveTypes.String, Name: "name", Required: false},
|
|
)
|
|
}
|