From 141413ad760ad3e8358f4fe0fbfe3e0162775f77 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 20 Apr 2026 22:48:01 -0700 Subject: [PATCH] fix(tests): make tests pass on 32-bit architectures (#9168) (#9170) Two separate failures reported on 32-bit builds (void-linux 4.21): - weed/server: errorStreamImpl.count (and the same pattern in slowStream plus local totalEventsSent/totalSends) was a bare int64 sitting after smaller fields, so on 386/ARMv7/mips32 it landed at a 4-byte-aligned offset and atomic.AddInt64 panicked with "unaligned 64-bit atomic operation". Switched the counters to atomic.Int64, which Go guarantees is 8-byte aligned on every architecture. - weed/plugin/worker/iceberg: three equality-delete tests fail on 32-bit because the upstream github.com/apache/iceberg-go declares manifestEntry.EqualityIDs as *[]int while the Iceberg Avro schema defines equality_ids as long, and hamba/avro refuses to map Go int onto Avro long when int is 32-bit. Not fixable in seaweedfs, so guard the affected tests with a t.Skip() when unsafe.Sizeof(int) < 8 until the upstream type is changed to []int32/[]int64. --- weed/plugin/worker/iceberg/exec_test.go | 3 + weed/plugin/worker/iceberg/platform_test.go | 22 +++++++ .../server/filer_grpc_server_sub_meta_test.go | 60 ++++++++++++------- 3 files changed, 62 insertions(+), 23 deletions(-) create mode 100644 weed/plugin/worker/iceberg/platform_test.go diff --git a/weed/plugin/worker/iceberg/exec_test.go b/weed/plugin/worker/iceberg/exec_test.go index 3c26cc95f..9d284eb41 100644 --- a/weed/plugin/worker/iceberg/exec_test.go +++ b/weed/plugin/worker/iceberg/exec_test.go @@ -2668,6 +2668,7 @@ func TestCompactDataFilesWithPositionDeletes(t *testing.T) { } func TestCompactDataFilesWithEqualityDeletes(t *testing.T) { + skipIfEqualityDeleteAvroUnsupported(t) fs, client := startFakeFiler(t) setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"} @@ -2820,6 +2821,7 @@ func TestCompactDataFilesApplyDeletesDisabled(t *testing.T) { } func TestCompactDataFilesWithMixedDeletes(t *testing.T) { + skipIfEqualityDeleteAvroUnsupported(t) fs, client := startFakeFiler(t) setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"} @@ -3553,6 +3555,7 @@ func TestRewritePositionDeleteFilesPreservesUnsupportedMultiTargetDeletes(t *tes } func TestRewritePositionDeleteFilesRebuildsMixedDeleteManifests(t *testing.T) { + skipIfEqualityDeleteAvroUnsupported(t) fs, client := startFakeFiler(t) setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"} diff --git a/weed/plugin/worker/iceberg/platform_test.go b/weed/plugin/worker/iceberg/platform_test.go new file mode 100644 index 000000000..3f49714c3 --- /dev/null +++ b/weed/plugin/worker/iceberg/platform_test.go @@ -0,0 +1,22 @@ +package iceberg + +import ( + "testing" + "unsafe" +) + +// skipIfEqualityDeleteAvroUnsupported skips the test when running on a 32-bit +// architecture. The upstream github.com/apache/iceberg-go library declares +// manifestEntry.EqualityIDs as *[]int, but the Iceberg Avro schema declares +// equality_ids as a long array. hamba/avro rejects []int for Avro long when +// int is 32-bit (386, arm, mips), causing these tests to fail through no +// fault of seaweedfs. Tracked upstream; remove this guard once iceberg-go +// switches the field to []int32/[]int64. +func skipIfEqualityDeleteAvroUnsupported(t *testing.T) { + t.Helper() + if unsafe.Sizeof(int(0)) < 8 { + t.Skip("equality-delete Avro decoding is broken on 32-bit platforms " + + "because github.com/apache/iceberg-go declares EqualityIDs as *[]int " + + "but the Iceberg schema requires Avro long") + } +} diff --git a/weed/server/filer_grpc_server_sub_meta_test.go b/weed/server/filer_grpc_server_sub_meta_test.go index 2165899a6..5a60c5761 100644 --- a/weed/server/filer_grpc_server_sub_meta_test.go +++ b/weed/server/filer_grpc_server_sub_meta_test.go @@ -14,16 +14,19 @@ import ( // slowStream simulates a gRPC stream with configurable per-Send latency. // It counts individual events including those packed inside batches. +// Atomic counters use atomic.Int64 so they stay 8-byte aligned on 32-bit +// architectures (386, ARMv7, mips32) where a bare int64 struct field is +// only 4-byte aligned and panics under atomic.AddInt64. type slowStream struct { + sends atomic.Int64 // number of stream.Send() calls + eventsSent atomic.Int64 // total events (1 + len(Events) per Send) sendDelay time.Duration - sends int64 // number of stream.Send() calls - eventsSent int64 // total events (1 + len(Events) per Send) } func (s *slowStream) Send(msg *filer_pb.SubscribeMetadataResponse) error { time.Sleep(s.sendDelay) - atomic.AddInt64(&s.sends, 1) - atomic.AddInt64(&s.eventsSent, 1+int64(len(msg.Events))) + s.sends.Add(1) + s.eventsSent.Add(1 + int64(len(msg.Events))) return nil } @@ -114,9 +117,9 @@ func TestPipelinedSenderThroughput(t *testing.T) { } elapsed := time.Since(start) - directRate = float64(stream.eventsSent) / elapsed.Seconds() + directRate = float64(stream.eventsSent.Load()) / elapsed.Seconds() t.Logf("direct: %d events %4d sends %v %6.0f events/sec", - stream.eventsSent, stream.sends, elapsed.Round(time.Millisecond), directRate) + stream.eventsSent.Load(), stream.sends.Load(), elapsed.Round(time.Millisecond), directRate) }) // --- Pipelined + batched (new behavior): file reads overlap with batched sends --- @@ -139,9 +142,9 @@ func TestPipelinedSenderThroughput(t *testing.T) { } elapsed := time.Since(start) - batchedRate = float64(stream.eventsSent) / elapsed.Seconds() + batchedRate = float64(stream.eventsSent.Load()) / elapsed.Seconds() t.Logf("pipelined+batch: %d events %4d sends %v %6.0f events/sec", - stream.eventsSent, stream.sends, elapsed.Round(time.Millisecond), batchedRate) + stream.eventsSent.Load(), stream.sends.Load(), elapsed.Round(time.Millisecond), batchedRate) }) if directRate > 0 { @@ -216,11 +219,13 @@ func TestBatchingAdaptive(t *testing.T) { } sender.Close() + sends := stream.sends.Load() + events := stream.eventsSent.Load() t.Logf("old events: %d events in %d sends (avg batch size: %.1f)", - stream.eventsSent, stream.sends, float64(stream.eventsSent)/float64(stream.sends)) + events, sends, float64(events)/float64(sends)) - if stream.sends >= int64(numEvents) { - t.Errorf("expected batching to reduce sends below %d, got %d", numEvents, stream.sends) + if sends >= int64(numEvents) { + t.Errorf("expected batching to reduce sends below %d, got %d", numEvents, sends) } }) @@ -233,24 +238,29 @@ func TestBatchingAdaptive(t *testing.T) { } sender.Close() + sends := stream.sends.Load() + events := stream.eventsSent.Load() t.Logf("recent events: %d events in %d sends (avg batch size: %.1f)", - stream.eventsSent, stream.sends, float64(stream.eventsSent)/float64(stream.sends)) + events, sends, float64(events)/float64(sends)) - if stream.sends != int64(numEvents) { - t.Errorf("expected 1:1 sends for recent events, got %d sends for %d events", stream.sends, numEvents) + if sends != int64(numEvents) { + t.Errorf("expected 1:1 sends for recent events, got %d sends for %d events", sends, numEvents) } }) } // errorStreamImpl is a metadataStreamSender that returns an error after N sends. +// count uses atomic.Int64 so it stays 8-byte aligned on 32-bit architectures +// (386, ARMv7, mips32) where a bare int64 struct field after smaller fields +// is only 4-byte aligned and panics under atomic.AddInt64. type errorStreamImpl struct { + count atomic.Int64 failAfter int err error - count int64 } func (s *errorStreamImpl) Send(msg *filer_pb.SubscribeMetadataResponse) error { - n := atomic.AddInt64(&s.count, 1) + n := s.count.Add(1) if int(n) > s.failAfter { return s.err } @@ -352,8 +362,8 @@ func TestPipelinedSingleVsParallelStreams(t *testing.T) { err = closeErr } elapsed = time.Since(start) - eventsSent = atomic.LoadInt64(&stream.eventsSent) - sends = atomic.LoadInt64(&stream.sends) + eventsSent = stream.eventsSent.Load() + sends = stream.sends.Load() return } @@ -370,7 +380,10 @@ func TestPipelinedSingleVsParallelStreams(t *testing.T) { var parallelRate float64 t.Run("10_pipelined_streams", func(t *testing.T) { - var totalEventsSent, totalSends int64 + // atomic.Int64 guarantees 8-byte alignment on 32-bit architectures where + // a local int64 variable's address is only 4-byte aligned and atomic + // 64-bit operations panic with "unaligned 64-bit atomic operation". + var totalEventsSent, totalSends atomic.Int64 var wg sync.WaitGroup start := time.Now() @@ -379,16 +392,17 @@ func TestPipelinedSingleVsParallelStreams(t *testing.T) { go func(files []logFile) { defer wg.Done() eventsSent, sends, _, _ := simulatePipeline(files) - atomic.AddInt64(&totalEventsSent, eventsSent) - atomic.AddInt64(&totalSends, sends) + totalEventsSent.Add(eventsSent) + totalSends.Add(sends) }(partitions[d]) } wg.Wait() elapsed := time.Since(start) - parallelRate = float64(totalEventsSent) / elapsed.Seconds() + totalEvents := totalEventsSent.Load() + parallelRate = float64(totalEvents) / elapsed.Seconds() t.Logf("%d streams: %5d events %4d sends %v %7.0f events/sec", - numDirs, totalEventsSent, totalSends, elapsed.Round(time.Millisecond), parallelRate) + numDirs, totalEvents, totalSends.Load(), elapsed.Round(time.Millisecond), parallelRate) }) if singleRate > 0 && parallelRate > 0 {