Message Queue: Add sql querying (#7185)

* feat: Phase 1 - Add SQL query engine foundation for MQ topics Implements core SQL infrastructure with metadata operations: New Components: - SQL parser integration using github.com/xwb1989/sqlparser - Query engine framework in weed/query/engine/ - Schema catalog mapping MQ topics to SQL tables - Interactive SQL CLI command 'weed sql' Supported Operations: - SHOW DATABASES (lists MQ namespaces) - SHOW TABLES (lists MQ topics) - SQL statement parsing and routing - Error handling and result formatting Key Design Decisions: - MQ namespaces ↔ SQL databases - MQ topics ↔ SQL tables - Parquet message storage ready for querying - Backward-compatible schema evolution support Testing: - Unit tests for core engine functionality - Command integration tests - Parse error handling validation Assumptions (documented in code): - All MQ messages stored in Parquet format - Schema evolution maintains backward compatibility - MySQL-compatible SQL syntax via sqlparser - Single-threaded usage per SQL session Next Phase: DDL operations (CREATE/ALTER/DROP TABLE) * feat: Phase 2 - Add DDL operations and real MQ broker integration Implements comprehensive DDL support for MQ topic management: New Components: - Real MQ broker connectivity via BrokerClient - CREATE TABLE → ConfigureTopic gRPC calls - DROP TABLE → DeleteTopic operations - DESCRIBE table → Schema introspection - SQL type mapping (SQL ↔ MQ schema types) Enhanced Features: - Live topic discovery from MQ broker - Fallback to cached/sample data when broker unavailable - MySQL-compatible DESCRIBE output - Schema validation and error handling - CREATE TABLE with column definitions Key Infrastructure: - broker_client.go: gRPC communication with MQ broker - sql_types.go: Bidirectional SQL/MQ type conversion - describe.go: Table schema introspection - Enhanced engine.go: Full DDL routing and execution Supported SQL Operations: ✅ SHOW DATABASES, SHOW TABLES (live + fallback) ✅ CREATE TABLE table_name (col1 INT, col2 VARCHAR(50), ...) ✅ DROP TABLE table_name ✅ DESCRIBE table_name / SHOW COLUMNS FROM table_name Known Limitations: - SQL parser issues with reserved keywords (e.g., 'timestamp') - Requires running MQ broker for full functionality - ALTER TABLE not yet implemented - DeleteTopic method needs broker-side implementation Architecture Decisions: - Broker discovery via filer lock mechanism (same as shell commands) - Graceful fallback when broker unavailable - ConfigureTopic for CREATE TABLE with 6 default partitions - Schema versioning ready for ALTER TABLE support Testing: - Unit tests updated with filer address parameter - Integration tests for DDL operations - Error handling for connection failures Next Phase: SELECT query execution with Parquet scanning * fix: Resolve High Priority TODOs - Real MQ Broker Integration ✅ COMPLETED HIGH PRIORITY TODOs: 🔧 **Real FilerClient Integration** (engine.go:131) - Implemented GetFilerClient() method in BrokerClient - Added filerClientImpl with full FilerClient interface compliance - Added AdjustedUrl() and GetDataCenter() methods - Real filerClient connection replaces nil fallback 🔧 **Partition Discovery via MQ Broker** (hybrid_message_scanner.go:116) - Added ListTopicPartitions() method using topic configuration - Implemented discoverTopicPartitions() in HybridMessageScanner - Reads actual partition count from BrokerPartitionAssignments - Generates proper partition ranges based on topic.PartitionCount 📋 **Technical Fixes:** - Fixed compilation errors with undefined variables - Proper error handling with filerClientErr variable - Corrected ConfigureTopicResponse field usage (BrokerPartitionAssignments vs PartitionCount) - Complete FilerClient interface implementation 🎯 **Impact:** - SQL engine now connects to real MQ broker infrastructure - Actual topic partition discovery instead of hardcoded defaults - Production-ready broker integration with graceful fallbacks - Maintains backward compatibility with sample data when broker unavailable ✅ All tests passing - High priority TODO resolution complete! Next: Schema-aware message parsing and time filter optimization. * feat: Time Filter Extraction - Complete Performance Optimization ✅ FOURTH HIGH PRIORITY TODO COMPLETED! ⏰ **Time Filter Extraction & Push-Down Optimization** (engine.go:198-199) - Replaced hardcoded StartTimeNs=0, StopTimeNs=0 with intelligent extraction - Added extractTimeFilters() with recursive WHERE clause analysis - Smart time column detection (\_timestamp_ns, created_at, timestamp, etc.) - Comprehensive time value parsing (nanoseconds, ISO dates, datetime formats) - Operator reversal handling (column op value vs value op column) 🧠 **Intelligent WHERE Clause Processing:** - AND expressions: Combine time bounds (intersection) ✅ - OR expressions: Skip extraction (safety) ✅ - Parentheses: Recursive unwrapping ✅ - Comparison operators: >, >=, <, <=, = ✅ - Multiple time formats: nanoseconds, RFC3339, date-only, datetime ✅ 🚀 **Performance Impact:** - Push-down filtering to hybrid scanner level - Reduced data scanning at source (live logs + Parquet files) - Time-based partition pruning potential - Significant performance gains for time-series queries 📊 **Comprehensive Testing (21 tests passing):** - ✅ Time filter extraction (6 test scenarios) - ✅ Time column recognition (case-insensitive) - ✅ Time value parsing (5 formats) - ✅ Full integration with SELECT queries - ✅ Backward compatibility maintained 💡 **Real-World Query Examples:** Before: Scans ALL data, filters in memory SELECT * FROM events WHERE \_timestamp_ns > 1672531200000000000; After: Scans ONLY relevant time range at source level → StartTimeNs=1672531200000000000, StopTimeNs=0 → Massive performance improvement for large datasets! 🎯 **Production Ready Features:** - Multiple time column formats supported - Graceful fallbacks for invalid dates - OR clause safety (avoids incorrect optimization) - Comprehensive error handling **ALL MEDIUM PRIORITY TODOs NOW READY FOR NEXT PHASEtest ./weed/query/engine/ -v* 🎉 * feat: Extended WHERE Operators - Complete Advanced Filtering ✅ **EXTENDED WHERE OPERATORS IMPLEMENTEDtest ./weed/query/engine/ -v | grep -E PASS * feat: Enhanced SQL CLI Experience ✅ COMPLETE ENHANCED CLI IMPLEMENTATION: 🚀 **Multiple Execution Modes:** - Interactive shell with enhanced prompts and context - Single query execution: --query 'SQL' --output format - Batch file processing: --file queries.sql --output csv - Database context switching: --database dbname 📊 **Multi-Format Output:** - Table format (ASCII) - default for interactive - JSON format - structured data for programmatic use - CSV format - spreadsheet-friendly output - Smart auto-detection based on execution mode ⚙️ **Enhanced Interactive Shell:** - Database context switching: USE database_name; - Output format switching: \format table|json|csv - Command history tracking (basic implementation) - Enhanced help with WHERE operator examples - Contextual prompts: seaweedfs:dbname> 🛠️ **Production Features:** - Comprehensive error handling (JSON + user-friendly) - Query execution timing and performance metrics - 30-second timeout protection with graceful handling - Real MQ integration with hybrid data scanning 📖 **Complete CLI Interface:** - Full flag support: --server, --interactive, --file, --output, --database, --query - Auto-detection of execution mode and output format - Structured help system with practical examples - Batch processing with multi-query file support 💡 **Advanced WHERE Integration:** All extended operators (<=, >=, !=, LIKE, IN) fully supported across all execution modes and output formats. 🎯 **Usage Examples:** - weed sql --interactive - weed sql --query 'SHOW DATABASES' --output json - weed sql --file queries.sql --output csv - weed sql --database analytics --interactive Enhanced CLI experience complete - production ready! 🚀 * Delete test_utils_test.go * fmt * integer conversion * show databases works * show tables works * Update describe.go * actual column types * Update .gitignore * scan topic messages * remove emoji * support aggregation functions * column name case insensitive, better auto column names * fmt * fix reading system fields * use parquet statistics for optimization * remove emoji * parquet file generate stats * scan all files * parquet file generation remember the sources also * fmt * sql * truncate topic * combine parquet results with live logs * explain * explain the execution plan * add tests * improve tests * skip * use mock for testing * add tests * refactor * fix after refactoring * detailed logs during explain. Fix bugs on reading live logs. * fix decoding data * save source buffer index start for log files * process buffer from brokers * filter out already flushed messages * dedup with buffer start index * explain with broker buffer * the parquet file should also remember the first buffer_start attribute from the sources * parquet file can query messages in broker memory, if log files do not exist * buffer start stored as 8 bytes * add jdbc * add postgres protocol * Revert "add jdbc" This reverts commit a6e48b76905d94e9c90953d6078660b4f038aa1e. * hook up seaweed sql engine * setup integration test for postgres * rename to "weed db" * return fast on error * fix versioning * address comments * address some comments * column name can be on left or right in where conditions * avoid sample data * remove sample data * de-support alter table and drop table * address comments * read broker, logs, and parquet files * Update engine.go * address some comments * use schema instead of inferred result types * fix tests * fix todo * fix empty spaces and coercion * fmt * change to pg_query_go * fix tests * fix tests * fmt * fix: Enable CGO in Docker build for pg_query_go dependency The pg_query_go library requires CGO to be enabled as it wraps the libpg_query C library. Added gcc and musl-dev dependencies to the Docker build for proper compilation. * feat: Replace pg_query_go with lightweight SQL parser (no CGO required) - Remove github.com/pganalyze/pg_query_go/v6 dependency to avoid CGO requirement - Implement lightweight SQL parser for basic SELECT, SHOW, and DDL statements - Fix operator precedence in WHERE clause parsing (handle AND/OR before comparisons) - Support INTEGER, FLOAT, and STRING literals in WHERE conditions - All SQL engine tests passing with new parser - PostgreSQL integration tests can now build without CGO The lightweight parser handles the essential SQL features needed for the SeaweedFS query engine while maintaining compatibility and avoiding CGO dependencies that caused Docker build issues. * feat: Add Parquet logical types to mq_schema.proto Added support for Parquet logical types in SeaweedFS message queue schema: - TIMESTAMP: UTC timestamp in microseconds since epoch with timezone flag - DATE: Date as days since Unix epoch (1970-01-01) - DECIMAL: Arbitrary precision decimal with configurable precision/scale - TIME: Time of day in microseconds since midnight These types enable advanced analytics features: - Time-based filtering and window functions - Date arithmetic and year/month/day extraction - High-precision numeric calculations - Proper time zone handling for global deployments Regenerated protobuf Go code with new scalar types and value messages. * feat: Enable publishers to use Parquet logical types Enhanced MQ publishers to utilize the new logical types: - Updated convertToRecordValue() to use TimestampValue instead of string RFC3339 - Added DateValue support for birth_date field (days since epoch) - Added DecimalValue support for precise_amount field with configurable precision/scale - Enhanced UserEvent struct with PreciseAmount and BirthDate fields - Added convertToDecimal() helper using big.Rat for precise decimal conversion - Updated test data generator to produce varied birth dates (1970-2005) and precise amounts Publishers now generate structured data with proper logical types: - ✅ TIMESTAMP: Microsecond precision UTC timestamps - ✅ DATE: Birth dates as days since Unix epoch - ✅ DECIMAL: Precise amounts with 18-digit precision, 4-decimal scale Successfully tested with PostgreSQL integration - all topics created with logical type data. * feat: Add logical type support to SQL query engine Extended SQL engine to handle new Parquet logical types: - Added TimestampValue comparison support (microsecond precision) - Added DateValue comparison support (days since epoch) - Added DecimalValue comparison support with string conversion - Added TimeValue comparison support (microseconds since midnight) - Enhanced valuesEqual(), valueLessThan(), valueGreaterThan() functions - Added decimalToString() helper for precise decimal-to-string conversion - Imported math/big for arbitrary precision decimal handling The SQL engine can now: - ✅ Compare TIMESTAMP values for filtering (e.g., WHERE timestamp > 1672531200000000000) - ✅ Compare DATE values for date-based queries (e.g., WHERE birth_date >= 12345) - ✅ Compare DECIMAL values for precise financial calculations - ✅ Compare TIME values for time-of-day filtering Next: Add YEAR(), MONTH(), DAY() extraction functions for date analytics. * feat: Add window function foundation with timestamp support Added comprehensive foundation for SQL window functions with timestamp analytics: Core Window Function Types: - WindowSpec with PartitionBy and OrderBy support - WindowFunction struct for ROW_NUMBER, RANK, LAG, LEAD - OrderByClause for timestamp-based ordering - Extended SelectStatement to support WindowFunctions field Timestamp Analytics Functions: ✅ ApplyRowNumber() - ROW_NUMBER() OVER (ORDER BY timestamp) ✅ ExtractYear() - Extract year from TIMESTAMP logical type ✅ ExtractMonth() - Extract month from TIMESTAMP logical type ✅ ExtractDay() - Extract day from TIMESTAMP logical type ✅ FilterByYear() - Filter records by timestamp year Foundation for Advanced Window Functions: - LAG/LEAD for time-series access to previous/next values - RANK/DENSE_RANK for temporal ranking - FIRST_VALUE/LAST_VALUE for window boundaries - PARTITION BY support for grouped analytics This enables sophisticated time-series analytics like: - SELECT *, ROW_NUMBER() OVER (ORDER BY timestamp) FROM user_events WHERE EXTRACT(YEAR FROM timestamp) = 2024 - Trend analysis over time windows - Session analytics with LAG/LEAD functions - Time-based ranking and percentiles Ready for production time-series analytics with proper timestamp logical type support! 🚀 * fmt * fix * fix describe issue * fix tests, avoid panic * no more mysql * timeout client connections * Update SQL_FEATURE_PLAN.md * handling errors * remove sleep * fix splitting multiple SQLs * fixes * fmt * fix * Update weed/util/log_buffer/log_buffer.go Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> * Update SQL_FEATURE_PLAN.md Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> * code reuse * fix * fix * feat: Add basic arithmetic operators (+, -, *, /, %) with comprehensive tests - Implement EvaluateArithmeticExpression with support for all basic operators - Handle type conversions between int, float, string, and boolean - Add proper error handling for division/modulo by zero - Include 14 comprehensive test cases covering all edge cases - Support mixed type arithmetic (int + float, string numbers, etc.) All tests passing ✅ * feat: Add mathematical functions ROUND, CEIL, FLOOR, ABS with comprehensive tests - Implement ROUND with optional precision parameter - Add CEIL function for rounding up to nearest integer - Add FLOOR function for rounding down to nearest integer - Add ABS function for absolute values with type preservation - Support all numeric types (int32, int64, float32, double) - Comprehensive test suite with 20+ test cases covering: - Positive/negative numbers - Integer/float type preservation - Precision handling for ROUND - Null value error handling - Edge cases (zero, large numbers) All tests passing ✅ * feat: Add date/time functions CURRENT_DATE, CURRENT_TIMESTAMP, EXTRACT with comprehensive tests - Implement CURRENT_DATE returning YYYY-MM-DD format - Add CURRENT_TIMESTAMP returning TimestampValue with microseconds - Add CURRENT_TIME returning HH:MM:SS format - Add NOW() as alias for CURRENT_TIMESTAMP - Implement comprehensive EXTRACT function supporting: - YEAR, MONTH, DAY, HOUR, MINUTE, SECOND - QUARTER, WEEK, DOY (day of year), DOW (day of week) - EPOCH (Unix timestamp) - Support multiple input formats: - TimestampValue (microseconds) - String dates (multiple formats) - Unix timestamps (int64 seconds) - Comprehensive test suite with 15+ test cases covering: - All date/time constants - Extract from different value types - Error handling for invalid inputs - Timezone handling All tests passing ✅ * feat: Add DATE_TRUNC function with comprehensive tests - Implement comprehensive DATE_TRUNC function supporting: - Time precisions: microsecond, millisecond, second, minute, hour - Date precisions: day, week, month, quarter, year, decade, century, millennium - Support both singular and plural forms (e.g., 'minute' and 'minutes') - Enhanced date/time parsing with proper timezone handling: - Assume local timezone for non-timezone string formats - Support UTC formats with explicit timezone indicators - Consistent behavior between parsing and truncation - Comprehensive test suite with 11 test cases covering: - All supported precisions from microsecond to year - Multiple input types (TimestampValue, string dates) - Edge cases (null values, invalid precisions) - Timezone consistency validation All tests passing ✅ * feat: Add comprehensive string functions with extensive tests Implemented String Functions: - LENGTH: Get string length (supports all value types) - UPPER/LOWER: Case conversion - TRIM/LTRIM/RTRIM: Whitespace removal (space, tab, newline, carriage return) - SUBSTRING: Extract substring with optional length (SQL 1-based indexing) - CONCAT: Concatenate multiple values (supports mixed types, skips nulls) - REPLACE: Replace all occurrences of substring - POSITION: Find substring position (1-based, 0 if not found) - LEFT/RIGHT: Extract leftmost/rightmost characters - REVERSE: Reverse string with proper Unicode support Key Features: - Robust type conversion (string, int, float, bool, bytes) - Unicode-safe operations (proper rune handling in REVERSE) - SQL-compatible indexing (1-based for SUBSTRING, POSITION) - Comprehensive error handling with descriptive messages - Mixed-type support (e.g., CONCAT number with string) Helper Functions: - valueToString: Convert any schema_pb.Value to string - valueToInt64: Convert numeric values to int64 Comprehensive test suite with 25+ test cases covering: - All string functions with typical use cases - Type conversion scenarios (numbers, booleans) - Edge cases (empty strings, null values, Unicode) - Error conditions and boundary testing All tests passing ✅ * refactor: Split sql_functions.go into smaller, focused files **File Structure Before:** - sql_functions.go (850+ lines) - sql_functions_test.go (1,205+ lines) **File Structure After:** - function_helpers.go (105 lines) - shared utility functions - arithmetic_functions.go (205 lines) - arithmetic operators & math functions - datetime_functions.go (170 lines) - date/time functions & constants - string_functions.go (335 lines) - string manipulation functions - arithmetic_functions_test.go (560 lines) - tests for arithmetic & math - datetime_functions_test.go (370 lines) - tests for date/time functions - string_functions_test.go (270 lines) - tests for string functions **Benefits:** ✅ Better organization by functional domain ✅ Easier to find and maintain specific function types ✅ Smaller, more manageable file sizes ✅ Clear separation of concerns ✅ Improved code readability and navigation ✅ All tests passing - no functionality lost **Total:** 7 focused files (1,455 lines) vs 2 monolithic files (2,055+ lines) This refactoring improves maintainability while preserving all functionality. * fix: Improve test stability for date/time functions **Problem:** - CURRENT_TIMESTAMP test had timing race condition that could cause flaky failures - CURRENT_DATE test could fail if run exactly at midnight boundary - Tests were too strict about timing precision without accounting for system variations **Root Cause:** - Test captured before/after timestamps and expected function result to be exactly between them - No tolerance for clock precision differences, NTP adjustments, or system timing variations - Date boundary race condition around midnight transitions **Solution:** ✅ **CURRENT_TIMESTAMP test**: Added 100ms tolerance buffer to account for: - Clock precision differences between time.Now() calls - System timing variations and NTP corrections - Microsecond vs nanosecond precision differences ✅ **CURRENT_DATE test**: Enhanced to handle midnight boundary crossings: - Captures date before and after function call - Accepts either date value in case of midnight transition - Prevents false failures during overnight test runs **Testing:** - Verified with repeated test runs (5x iterations) - all pass consistently - Full test suite passes - no regressions introduced - Tests are now robust against timing edge cases **Impact:** 🚀 **Eliminated flaky test failures** while maintaining function correctness validation 🔧 **Production-ready testing** that works across different system environments ⚡ **CI/CD reliability** - tests won't fail due to timing variations * heap sort the data sources * int overflow * Update README.md * redirect GetUnflushedMessages to brokers hosting the topic partition * Update postgres-examples/README.md Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> * clean up * support limit with offset * Update SQL_FEATURE_PLAN.md * limit with offset * ensure int conversion correctness * Update weed/query/engine/hybrid_message_scanner.go Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> * avoid closing closed channel * support string concatenation || * int range * using consts; avoid test data in production binary * fix tests * Update SQL_FEATURE_PLAN.md * fix "use db" * address comments * fix comments * Update mocks_test.go * comment * improve docker build * normal if no partitions found * fix build docker * Update SQL_FEATURE_PLAN.md * upgrade to raft v1.1.4 resolving race in leader * raft 1.1.5 * Update SQL_FEATURE_PLAN.md * Revert "raft 1.1.5" This reverts commit 5f3bdfadbfd50daa5733b72cf09f17d4bfb79ee6. * Revert "upgrade to raft v1.1.4 resolving race in leader" This reverts commit fa620f0223ce02b59e96d94a898c2ad9464657d2. * Fix data race in FUSE GetAttr operation - Add shared lock to GetAttr when accessing file handle entries - Prevents concurrent access between Write (ExclusiveLock) and GetAttr (SharedLock) - Fixes race on entry.Attributes.FileSize field during concurrent operations - Write operations already use ExclusiveLock, now GetAttr uses SharedLock for consistency Resolves race condition: Write at weedfs_file_write.go:62 vs Read at filechunks.go:28 * Update weed/mq/broker/broker_grpc_query.go Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> * clean up * Update db.go * limit with offset * Update Makefile * fix id*2 * fix math * fix string function bugs and add tests * fix string concat * ensure empty spaces for literals * add ttl for catalog * fix time functions * unused code path * database qualifier * refactor * extract * recursive functions * add cockroachdb parser * postgres only * test SQLs * fix tests * fix count * * fix where clause * fix limit offset * fix count fast path * fix tests * func name * fix database qualifier * fix tests * Update engine.go * fix tests * fix jaeger https://github.com/advisories/GHSA-2w8w-qhg4-f78j * remove order by, group by, join * fix extract * prevent single quote in the string * skip control messages * skip control message when converting to parquet files * psql change database * remove old code * remove old parser code * rename file * use db * fix alias * add alias test * compare int64 * fix _timestamp_ns comparing * alias support * fix fast path count * rendering data sources tree * reading data sources * reading parquet logic types * convert logic types to parquet * go mod * fmt * skip decimal types * use UTC * add warning if broker fails * add user password file * support IN * support INTERVAL * _ts as timestamp column * _ts can compare with string * address comments * is null / is not null * go mod * clean up * restructure execution plan * remove extra double quotes * fix converting logical types to parquet * decimal * decimal support * do not skip decimal logical types * making row-building schema-aware and alignment-safe Emit parquet.NullValue() for missing fields to keep row shapes aligned. Always advance list level and safely handle nil list values. Add toParquetValueForType(...) to coerce values to match the declared Parquet type (e.g., STRING/BYTES via byte array; numeric/string conversions for INT32/INT64/DOUBLE/FLOAT/BOOL/TIMESTAMP/DATE/TIME). Keep nil-byte guards for ByteArray. * tests for growslice * do not batch * live logs in sources can be skipped in execution plan * go mod tidy * Update fuse-integration.yml * Update Makefile * fix deprecated * fix deprecated * remove deep-clean all rows * broker memory count * fix FieldIndex --------- Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
2025-09-09 01:01:03 -07:00
parent 30d69fa778
commit a7fdc0d137
117 changed files with 33189 additions and 367 deletions
--- a/weed/mq/broker/broker_grpc_pub.go
+++ b/weed/mq/broker/broker_grpc_pub.go
@@ -12,7 +12,9 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/mq/topic"
 	"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/schema_pb"
 	"google.golang.org/grpc/peer"
+	"google.golang.org/protobuf/proto"
 )

 // PUB
@@ -140,6 +142,16 @@ func (b *MessageQueueBroker) PublishMessage(stream mq_pb.SeaweedMessaging_Publis
 			continue
 		}

+		// Basic validation: ensure message can be unmarshaled as RecordValue
+		if dataMessage.Value != nil {
+			record := &schema_pb.RecordValue{}
+			if err := proto.Unmarshal(dataMessage.Value, record); err == nil {
+			} else {
+				// If unmarshaling fails, we skip validation but log a warning
+				glog.V(1).Infof("Could not unmarshal RecordValue for validation on topic %v partition %v: %v", initMessage.Topic, initMessage.Partition, err)
+			}
+		}
+
 		// The control message should still be sent to the follower
 		// to avoid timing issue when ack messages.

@@ -171,3 +183,4 @@ func findClientAddress(ctx context.Context) string {
 	}
 	return pr.Addr.String()
 }
+
--- a/weed/mq/broker/broker_grpc_query.go
+++ b/weed/mq/broker/broker_grpc_query.go
@@ -0,0 +1,358 @@
+package broker
+
+import (
+	"context"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"io"
+	"strings"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/mq/topic"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/schema_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util/log_buffer"
+)
+
+// BufferRange represents a range of buffer indexes that have been flushed to disk
+type BufferRange struct {
+	start int64
+	end   int64
+}
+
+// ErrNoPartitionAssignment indicates no broker assignment found for the partition.
+// This is a normal case that means there are no unflushed messages for this partition.
+var ErrNoPartitionAssignment = errors.New("no broker assignment found for partition")
+
+// GetUnflushedMessages returns messages from the broker's in-memory LogBuffer
+// that haven't been flushed to disk yet, using buffer_start metadata for deduplication
+// Now supports streaming responses and buffer index filtering for better performance
+// Includes broker routing to redirect requests to the correct broker hosting the topic/partition
+func (b *MessageQueueBroker) GetUnflushedMessages(req *mq_pb.GetUnflushedMessagesRequest, stream mq_pb.SeaweedMessaging_GetUnflushedMessagesServer) error {
+	// Convert protobuf types to internal types
+	t := topic.FromPbTopic(req.Topic)
+	partition := topic.FromPbPartition(req.Partition)
+
+	glog.V(2).Infof("GetUnflushedMessages request for %v %v", t, partition)
+
+	// Get the local partition for this topic/partition
+	b.accessLock.Lock()
+	localPartition := b.localTopicManager.GetLocalPartition(t, partition)
+	b.accessLock.Unlock()
+
+	if localPartition == nil {
+		// Topic/partition not found locally, attempt to find the correct broker and redirect
+		glog.V(1).Infof("Topic/partition %v %v not found locally, looking up broker", t, partition)
+
+		// Look up which broker hosts this topic/partition
+		brokerHost, err := b.findBrokerForTopicPartition(req.Topic, req.Partition)
+		if err != nil {
+			if errors.Is(err, ErrNoPartitionAssignment) {
+				// Normal case: no broker assignment means no unflushed messages
+				glog.V(2).Infof("No broker assignment for %v %v - no unflushed messages", t, partition)
+				return stream.Send(&mq_pb.GetUnflushedMessagesResponse{
+					EndOfStream: true,
+				})
+			}
+			return stream.Send(&mq_pb.GetUnflushedMessagesResponse{
+				Error:       fmt.Sprintf("failed to find broker for %v %v: %v", t, partition, err),
+				EndOfStream: true,
+			})
+		}
+
+		if brokerHost == "" {
+			// This should not happen after ErrNoPartitionAssignment check, but keep for safety
+			glog.V(2).Infof("Empty broker host for %v %v - no unflushed messages", t, partition)
+			return stream.Send(&mq_pb.GetUnflushedMessagesResponse{
+				EndOfStream: true,
+			})
+		}
+
+		// Redirect to the correct broker
+		glog.V(1).Infof("Redirecting GetUnflushedMessages request for %v %v to broker %s", t, partition, brokerHost)
+		return b.redirectGetUnflushedMessages(brokerHost, req, stream)
+	}
+
+	// Build deduplication map from existing log files using buffer_start metadata
+	partitionDir := topic.PartitionDir(t, partition)
+	flushedBufferRanges, err := b.buildBufferStartDeduplicationMap(partitionDir)
+	if err != nil {
+		glog.Errorf("Failed to build deduplication map for %v %v: %v", t, partition, err)
+		// Continue with empty map - better to potentially duplicate than to miss data
+		flushedBufferRanges = make([]BufferRange, 0)
+	}
+
+	// Use buffer_start index for precise deduplication
+	lastFlushTsNs := localPartition.LogBuffer.LastFlushTsNs
+	startBufferIndex := req.StartBufferIndex
+	startTimeNs := lastFlushTsNs // Still respect last flush time for safety
+
+	glog.V(2).Infof("Streaming unflushed messages for %v %v, buffer >= %d, timestamp >= %d (safety), excluding %d flushed buffer ranges",
+		t, partition, startBufferIndex, startTimeNs, len(flushedBufferRanges))
+
+	// Stream messages from LogBuffer with filtering
+	messageCount := 0
+	startPosition := log_buffer.NewMessagePosition(startTimeNs, startBufferIndex)
+
+	// Use the new LoopProcessLogDataWithBatchIndex method to avoid code duplication
+	_, _, err = localPartition.LogBuffer.LoopProcessLogDataWithBatchIndex(
+		"GetUnflushedMessages",
+		startPosition,
+		0,                            // stopTsNs = 0 means process all available data
+		func() bool { return false }, // waitForDataFn = false means don't wait for new data
+		func(logEntry *filer_pb.LogEntry, batchIndex int64) (isDone bool, err error) {
+			// Apply buffer index filtering if specified
+			if startBufferIndex > 0 && batchIndex < startBufferIndex {
+				glog.V(3).Infof("Skipping message from buffer index %d (< %d)", batchIndex, startBufferIndex)
+				return false, nil
+			}
+
+			// Check if this message is from a buffer range that's already been flushed
+			if b.isBufferIndexFlushed(batchIndex, flushedBufferRanges) {
+				glog.V(3).Infof("Skipping message from flushed buffer index %d", batchIndex)
+				return false, nil
+			}
+
+			// Stream this message
+			err = stream.Send(&mq_pb.GetUnflushedMessagesResponse{
+				Message: &mq_pb.LogEntry{
+					TsNs:             logEntry.TsNs,
+					Key:              logEntry.Key,
+					Data:             logEntry.Data,
+					PartitionKeyHash: uint32(logEntry.PartitionKeyHash),
+				},
+				EndOfStream: false,
+			})
+
+			if err != nil {
+				glog.Errorf("Failed to stream message: %v", err)
+				return true, err // isDone = true to stop processing
+			}
+
+			messageCount++
+			return false, nil // Continue processing
+		},
+	)
+
+	// Handle collection errors
+	if err != nil && err != log_buffer.ResumeFromDiskError {
+		streamErr := stream.Send(&mq_pb.GetUnflushedMessagesResponse{
+			Error:       fmt.Sprintf("failed to stream unflushed messages: %v", err),
+			EndOfStream: true,
+		})
+		if streamErr != nil {
+			glog.Errorf("Failed to send error response: %v", streamErr)
+		}
+		return err
+	}
+
+	// Send end-of-stream marker
+	err = stream.Send(&mq_pb.GetUnflushedMessagesResponse{
+		EndOfStream: true,
+	})
+
+	if err != nil {
+		glog.Errorf("Failed to send end-of-stream marker: %v", err)
+		return err
+	}
+
+	glog.V(1).Infof("Streamed %d unflushed messages for %v %v", messageCount, t, partition)
+	return nil
+}
+
+// buildBufferStartDeduplicationMap scans log files to build a map of buffer ranges
+// that have been flushed to disk, using the buffer_start metadata
+func (b *MessageQueueBroker) buildBufferStartDeduplicationMap(partitionDir string) ([]BufferRange, error) {
+	var flushedRanges []BufferRange
+
+	// List all files in the partition directory using filer client accessor
+	// Use pagination to handle directories with more than 1000 files
+	err := b.fca.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		var lastFileName string
+		var hasMore = true
+
+		for hasMore {
+			var currentBatchProcessed int
+			err := filer_pb.SeaweedList(context.Background(), client, partitionDir, "", func(entry *filer_pb.Entry, isLast bool) error {
+				currentBatchProcessed++
+				hasMore = !isLast // If this is the last entry of a full batch, there might be more
+				lastFileName = entry.Name
+
+				if entry.IsDirectory {
+					return nil
+				}
+
+				// Skip Parquet files - they don't represent buffer ranges
+				if strings.HasSuffix(entry.Name, ".parquet") {
+					return nil
+				}
+
+				// Skip offset files
+				if strings.HasSuffix(entry.Name, ".offset") {
+					return nil
+				}
+
+				// Get buffer start for this file
+				bufferStart, err := b.getLogBufferStartFromFile(entry)
+				if err != nil {
+					glog.V(2).Infof("Failed to get buffer start from file %s: %v", entry.Name, err)
+					return nil // Continue with other files
+				}
+
+				if bufferStart == nil {
+					// File has no buffer metadata - skip deduplication for this file
+					glog.V(2).Infof("File %s has no buffer_start metadata", entry.Name)
+					return nil
+				}
+
+				// Calculate the buffer range covered by this file
+				chunkCount := int64(len(entry.GetChunks()))
+				if chunkCount > 0 {
+					fileRange := BufferRange{
+						start: bufferStart.StartIndex,
+						end:   bufferStart.StartIndex + chunkCount - 1,
+					}
+					flushedRanges = append(flushedRanges, fileRange)
+					glog.V(3).Infof("File %s covers buffer range [%d-%d]", entry.Name, fileRange.start, fileRange.end)
+				}
+
+				return nil
+			}, lastFileName, false, 1000) // Start from last processed file name for next batch
+
+			if err != nil {
+				return err
+			}
+
+			// If we processed fewer than 1000 entries, we've reached the end
+			if currentBatchProcessed < 1000 {
+				hasMore = false
+			}
+		}
+
+		return nil
+	})
+
+	if err != nil {
+		return flushedRanges, fmt.Errorf("failed to list partition directory %s: %v", partitionDir, err)
+	}
+
+	return flushedRanges, nil
+}
+
+// getLogBufferStartFromFile extracts LogBufferStart metadata from a log file
+func (b *MessageQueueBroker) getLogBufferStartFromFile(entry *filer_pb.Entry) (*LogBufferStart, error) {
+	if entry.Extended == nil {
+		return nil, nil
+	}
+
+	// Only support binary buffer_start format
+	if startData, exists := entry.Extended["buffer_start"]; exists {
+		if len(startData) == 8 {
+			startIndex := int64(binary.BigEndian.Uint64(startData))
+			if startIndex > 0 {
+				return &LogBufferStart{StartIndex: startIndex}, nil
+			}
+		} else {
+			return nil, fmt.Errorf("invalid buffer_start format: expected 8 bytes, got %d", len(startData))
+		}
+	}
+
+	return nil, nil
+}
+
+// isBufferIndexFlushed checks if a buffer index is covered by any of the flushed ranges
+func (b *MessageQueueBroker) isBufferIndexFlushed(bufferIndex int64, flushedRanges []BufferRange) bool {
+	for _, flushedRange := range flushedRanges {
+		if bufferIndex >= flushedRange.start && bufferIndex <= flushedRange.end {
+			return true
+		}
+	}
+	return false
+}
+
+// findBrokerForTopicPartition finds which broker hosts the specified topic/partition
+func (b *MessageQueueBroker) findBrokerForTopicPartition(topic *schema_pb.Topic, partition *schema_pb.Partition) (string, error) {
+	// Use LookupTopicBrokers to find which broker hosts this topic/partition
+	ctx := context.Background()
+	lookupReq := &mq_pb.LookupTopicBrokersRequest{
+		Topic: topic,
+	}
+
+	// If we're not the lock owner (balancer), we need to redirect to the balancer first
+	var lookupResp *mq_pb.LookupTopicBrokersResponse
+	var err error
+
+	if !b.isLockOwner() {
+		// Redirect to balancer to get topic broker assignments
+		balancerAddress := pb.ServerAddress(b.lockAsBalancer.LockOwner())
+		err = b.withBrokerClient(false, balancerAddress, func(client mq_pb.SeaweedMessagingClient) error {
+			lookupResp, err = client.LookupTopicBrokers(ctx, lookupReq)
+			return err
+		})
+	} else {
+		// We are the balancer, handle the lookup directly
+		lookupResp, err = b.LookupTopicBrokers(ctx, lookupReq)
+	}
+
+	if err != nil {
+		return "", fmt.Errorf("failed to lookup topic brokers: %v", err)
+	}
+
+	// Find the broker assignment that matches our partition
+	for _, assignment := range lookupResp.BrokerPartitionAssignments {
+		if b.partitionsMatch(partition, assignment.Partition) {
+			if assignment.LeaderBroker != "" {
+				return assignment.LeaderBroker, nil
+			}
+		}
+	}
+
+	return "", ErrNoPartitionAssignment
+}
+
+// partitionsMatch checks if two partitions represent the same partition
+func (b *MessageQueueBroker) partitionsMatch(p1, p2 *schema_pb.Partition) bool {
+	return p1.RingSize == p2.RingSize &&
+		p1.RangeStart == p2.RangeStart &&
+		p1.RangeStop == p2.RangeStop &&
+		p1.UnixTimeNs == p2.UnixTimeNs
+}
+
+// redirectGetUnflushedMessages forwards the GetUnflushedMessages request to the correct broker
+func (b *MessageQueueBroker) redirectGetUnflushedMessages(brokerHost string, req *mq_pb.GetUnflushedMessagesRequest, stream mq_pb.SeaweedMessaging_GetUnflushedMessagesServer) error {
+	ctx := stream.Context()
+
+	// Connect to the target broker and forward the request
+	return b.withBrokerClient(false, pb.ServerAddress(brokerHost), func(client mq_pb.SeaweedMessagingClient) error {
+		// Create a new stream to the target broker
+		targetStream, err := client.GetUnflushedMessages(ctx, req)
+		if err != nil {
+			return fmt.Errorf("failed to create stream to broker %s: %v", brokerHost, err)
+		}
+
+		// Forward all responses from the target broker to our client
+		for {
+			response, err := targetStream.Recv()
+			if err != nil {
+				if errors.Is(err, io.EOF) {
+					// Normal end of stream
+					return nil
+				}
+				return fmt.Errorf("error receiving from broker %s: %v", brokerHost, err)
+			}
+
+			// Forward the response to our client
+			if sendErr := stream.Send(response); sendErr != nil {
+				return fmt.Errorf("error forwarding response to client: %v", sendErr)
+			}
+
+			// Check if this is the end of stream
+			if response.EndOfStream {
+				return nil
+			}
+		}
+	})
+}
--- a/weed/mq/broker/broker_server.go
+++ b/weed/mq/broker/broker_server.go
@@ -2,13 +2,14 @@ package broker

 import (
 	"context"
+	"sync"
+	"time"
+
 	"github.com/seaweedfs/seaweedfs/weed/filer_client"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/mq/pub_balancer"
 	"github.com/seaweedfs/seaweedfs/weed/mq/sub_coordinator"
 	"github.com/seaweedfs/seaweedfs/weed/mq/topic"
-	"sync"
-	"time"

 	"github.com/seaweedfs/seaweedfs/weed/cluster"
 	"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
--- a/weed/mq/broker/broker_topic_partition_read_write.go
+++ b/weed/mq/broker/broker_topic_partition_read_write.go
@@ -2,13 +2,21 @@ package broker

 import (
 	"fmt"
+	"sync/atomic"
+	"time"
+
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/mq/topic"
 	"github.com/seaweedfs/seaweedfs/weed/util/log_buffer"
-	"sync/atomic"
-	"time"
 )

+// LogBufferStart tracks the starting buffer index for a live log file
+// Buffer indexes are monotonically increasing, count = number of chunks
+// Now stored in binary format for efficiency
+type LogBufferStart struct {
+	StartIndex int64 // Starting buffer index (count = len(chunks))
+}
+
 func (b *MessageQueueBroker) genLogFlushFunc(t topic.Topic, p topic.Partition) log_buffer.LogFlushFuncType {
 	partitionDir := topic.PartitionDir(t, p)

@@ -21,10 +29,11 @@ func (b *MessageQueueBroker) genLogFlushFunc(t topic.Topic, p topic.Partition) l

 		targetFile := fmt.Sprintf("%s/%s", partitionDir, startTime.Format(topic.TIME_FORMAT))

-		// TODO append block with more metadata
+		// Get buffer index (now globally unique across restarts)
+		bufferIndex := logBuffer.GetBatchIndex()

 		for {
-			if err := b.appendToFile(targetFile, buf); err != nil {
+			if err := b.appendToFileWithBufferIndex(targetFile, buf, bufferIndex); err != nil {
 				glog.V(0).Infof("metadata log write failed %s: %v", targetFile, err)
 				time.Sleep(737 * time.Millisecond)
 			} else {
@@ -40,6 +49,6 @@ func (b *MessageQueueBroker) genLogFlushFunc(t topic.Topic, p topic.Partition) l
 			localPartition.NotifyLogFlushed(logBuffer.LastFlushTsNs)
 		}

-		glog.V(0).Infof("flushing at %d to %s size %d", logBuffer.LastFlushTsNs, targetFile, len(buf))
+		glog.V(0).Infof("flushing at %d to %s size %d from buffer %s (index %d)", logBuffer.LastFlushTsNs, targetFile, len(buf), logBuffer.GetName(), bufferIndex)
 	}
 }
--- a/weed/mq/broker/broker_write.go
+++ b/weed/mq/broker/broker_write.go
@@ -2,16 +2,23 @@ package broker

 import (
 	"context"
+	"encoding/binary"
 	"fmt"
+	"os"
+	"time"
+
 	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/operation"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/util"
-	"os"
-	"time"
 )

 func (b *MessageQueueBroker) appendToFile(targetFile string, data []byte) error {
+	return b.appendToFileWithBufferIndex(targetFile, data, 0)
+}
+
+func (b *MessageQueueBroker) appendToFileWithBufferIndex(targetFile string, data []byte, bufferIndex int64) error {

 	fileId, uploadResult, err2 := b.assignAndUpload(targetFile, data)
 	if err2 != nil {
@@ -35,10 +42,48 @@ func (b *MessageQueueBroker) appendToFile(targetFile string, data []byte) error
 				Gid:      uint32(os.Getgid()),
 			},
 		}
+
+		// Add buffer start index for deduplication tracking (binary format)
+		if bufferIndex != 0 {
+			entry.Extended = make(map[string][]byte)
+			bufferStartBytes := make([]byte, 8)
+			binary.BigEndian.PutUint64(bufferStartBytes, uint64(bufferIndex))
+			entry.Extended["buffer_start"] = bufferStartBytes
+		}
 	} else if err != nil {
 		return fmt.Errorf("find %s: %v", fullpath, err)
 	} else {
 		offset = int64(filer.TotalSize(entry.GetChunks()))
+
+		// Verify buffer index continuity for existing files (append operations)
+		if bufferIndex != 0 {
+			if entry.Extended == nil {
+				entry.Extended = make(map[string][]byte)
+			}
+
+			// Check for existing buffer start (binary format)
+			if existingData, exists := entry.Extended["buffer_start"]; exists {
+				if len(existingData) == 8 {
+					existingStartIndex := int64(binary.BigEndian.Uint64(existingData))
+
+					// Verify that the new buffer index is consecutive
+					// Expected index = start + number of existing chunks
+					expectedIndex := existingStartIndex + int64(len(entry.GetChunks()))
+					if bufferIndex != expectedIndex {
+						// This shouldn't happen in normal operation
+						// Log warning but continue (don't crash the system)
+						glog.Warningf("non-consecutive buffer index for %s. Expected %d, got %d",
+							fullpath, expectedIndex, bufferIndex)
+					}
+					// Note: We don't update the start index - it stays the same
+				}
+			} else {
+				// No existing buffer start, create new one (shouldn't happen for existing files)
+				bufferStartBytes := make([]byte, 8)
+				binary.BigEndian.PutUint64(bufferStartBytes, uint64(bufferIndex))
+				entry.Extended["buffer_start"] = bufferStartBytes
+			}
+		}
 	}

 	// append to existing chunks
--- a/weed/mq/logstore/log_to_parquet.go
+++ b/weed/mq/logstore/log_to_parquet.go
@@ -3,7 +3,13 @@ package logstore
 import (
 	"context"
 	"encoding/binary"
+	"encoding/json"
 	"fmt"
+	"io"
+	"os"
+	"strings"
+	"time"
+
 	"github.com/parquet-go/parquet-go"
 	"github.com/parquet-go/parquet-go/compress/zstd"
 	"github.com/seaweedfs/seaweedfs/weed/filer"
@@ -16,10 +22,6 @@ import (
 	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
 	"github.com/seaweedfs/seaweedfs/weed/util/log_buffer"
 	"google.golang.org/protobuf/proto"
-	"io"
-	"os"
-	"strings"
-	"time"
 )

 const (
@@ -217,25 +219,29 @@ func writeLogFilesToParquet(filerClient filer_pb.FilerClient, partitionDir strin
 		os.Remove(tempFile.Name())
 	}()

-	writer := parquet.NewWriter(tempFile, parquetSchema, parquet.Compression(&zstd.Codec{Level: zstd.DefaultLevel}))
+	// Enable column statistics for fast aggregation queries
+	writer := parquet.NewWriter(tempFile, parquetSchema,
+		parquet.Compression(&zstd.Codec{Level: zstd.DefaultLevel}),
+		parquet.DataPageStatistics(true), // Enable column statistics
+	)
 	rowBuilder := parquet.NewRowBuilder(parquetSchema)

 	var startTsNs, stopTsNs int64

 	for _, logFile := range logFileGroups {
-		fmt.Printf("compact %s/%s ", partitionDir, logFile.Name)
 		var rows []parquet.Row
 		if err := iterateLogEntries(filerClient, logFile, func(entry *filer_pb.LogEntry) error {

+			// Skip control entries without actual data (same logic as read operations)
+			if isControlEntry(entry) {
+				return nil
+			}
+
 			if startTsNs == 0 {
 				startTsNs = entry.TsNs
 			}
 			stopTsNs = entry.TsNs

-			if len(entry.Key) == 0 {
-				return nil
-			}
-
 			// write to parquet file
 			rowBuilder.Reset()

@@ -244,14 +250,25 @@ func writeLogFilesToParquet(filerClient filer_pb.FilerClient, partitionDir strin
 				return fmt.Errorf("unmarshal record value: %w", err)
 			}

+			// Initialize Fields map if nil (prevents nil map assignment panic)
+			if record.Fields == nil {
+				record.Fields = make(map[string]*schema_pb.Value)
+			}
+
 			record.Fields[SW_COLUMN_NAME_TS] = &schema_pb.Value{
 				Kind: &schema_pb.Value_Int64Value{
 					Int64Value: entry.TsNs,
 				},
 			}
+
+			// Handle nil key bytes to prevent growslice panic in parquet-go
+			keyBytes := entry.Key
+			if keyBytes == nil {
+				keyBytes = []byte{} // Use empty slice instead of nil
+			}
 			record.Fields[SW_COLUMN_NAME_KEY] = &schema_pb.Value{
 				Kind: &schema_pb.Value_BytesValue{
-					BytesValue: entry.Key,
+					BytesValue: keyBytes,
 				},
 			}

@@ -259,7 +276,17 @@ func writeLogFilesToParquet(filerClient filer_pb.FilerClient, partitionDir strin
 				return fmt.Errorf("add record value: %w", err)
 			}

-			rows = append(rows, rowBuilder.Row())
+			// Build row and normalize any nil ByteArray values to empty slices
+			row := rowBuilder.Row()
+			for i, value := range row {
+				if value.Kind() == parquet.ByteArray {
+					if value.ByteArray() == nil {
+						row[i] = parquet.ByteArrayValue([]byte{})
+					}
+				}
+			}
+
+			rows = append(rows, row)

 			return nil

@@ -267,8 +294,9 @@ func writeLogFilesToParquet(filerClient filer_pb.FilerClient, partitionDir strin
 			return fmt.Errorf("iterate log entry %v/%v: %w", partitionDir, logFile.Name, err)
 		}

-		fmt.Printf("processed %d rows\n", len(rows))
+		// Nil ByteArray handling is done during row creation

+		// Write all rows in a single call
 		if _, err := writer.WriteRows(rows); err != nil {
 			return fmt.Errorf("write rows: %w", err)
 		}
@@ -280,7 +308,22 @@ func writeLogFilesToParquet(filerClient filer_pb.FilerClient, partitionDir strin

 	// write to parquet file to partitionDir
 	parquetFileName := fmt.Sprintf("%s.parquet", time.Unix(0, startTsNs).UTC().Format("2006-01-02-15-04-05"))
-	if err := saveParquetFileToPartitionDir(filerClient, tempFile, partitionDir, parquetFileName, preference, startTsNs, stopTsNs); err != nil {
+
+	// Collect source log file names and buffer_start metadata for deduplication
+	var sourceLogFiles []string
+	var earliestBufferStart int64
+	for _, logFile := range logFileGroups {
+		sourceLogFiles = append(sourceLogFiles, logFile.Name)
+
+		// Extract buffer_start from log file metadata
+		if bufferStart := getBufferStartFromLogFile(logFile); bufferStart > 0 {
+			if earliestBufferStart == 0 || bufferStart < earliestBufferStart {
+				earliestBufferStart = bufferStart
+			}
+		}
+	}
+
+	if err := saveParquetFileToPartitionDir(filerClient, tempFile, partitionDir, parquetFileName, preference, startTsNs, stopTsNs, sourceLogFiles, earliestBufferStart); err != nil {
 		return fmt.Errorf("save parquet file %s: %v", parquetFileName, err)
 	}

@@ -288,7 +331,7 @@ func writeLogFilesToParquet(filerClient filer_pb.FilerClient, partitionDir strin

 }

-func saveParquetFileToPartitionDir(filerClient filer_pb.FilerClient, sourceFile *os.File, partitionDir, parquetFileName string, preference *operation.StoragePreference, startTsNs, stopTsNs int64) error {
+func saveParquetFileToPartitionDir(filerClient filer_pb.FilerClient, sourceFile *os.File, partitionDir, parquetFileName string, preference *operation.StoragePreference, startTsNs, stopTsNs int64, sourceLogFiles []string, earliestBufferStart int64) error {
 	uploader, err := operation.NewUploader()
 	if err != nil {
 		return fmt.Errorf("new uploader: %w", err)
@@ -321,6 +364,19 @@ func saveParquetFileToPartitionDir(filerClient filer_pb.FilerClient, sourceFile
 	binary.BigEndian.PutUint64(maxTsBytes, uint64(stopTsNs))
 	entry.Extended["max"] = maxTsBytes

+	// Store source log files for deduplication (JSON-encoded list)
+	if len(sourceLogFiles) > 0 {
+		sourceLogFilesJson, _ := json.Marshal(sourceLogFiles)
+		entry.Extended["sources"] = sourceLogFilesJson
+	}
+
+	// Store earliest buffer_start for precise broker deduplication
+	if earliestBufferStart > 0 {
+		bufferStartBytes := make([]byte, 8)
+		binary.BigEndian.PutUint64(bufferStartBytes, uint64(earliestBufferStart))
+		entry.Extended["buffer_start"] = bufferStartBytes
+	}
+
 	for i := int64(0); i < chunkCount; i++ {
 		fileId, uploadResult, err, _ := uploader.UploadWithRetry(
 			filerClient,
@@ -362,7 +418,6 @@ func saveParquetFileToPartitionDir(filerClient filer_pb.FilerClient, sourceFile
 	}); err != nil {
 		return fmt.Errorf("create entry: %w", err)
 	}
-	fmt.Printf("saved to %s/%s\n", partitionDir, parquetFileName)

 	return nil
 }
@@ -389,7 +444,6 @@ func eachFile(entry *filer_pb.Entry, lookupFileIdFn func(ctx context.Context, fi
 			continue
 		}
 		if chunk.IsChunkManifest {
-			fmt.Printf("this should not happen. unexpected chunk manifest in %s", entry.Name)
 			return
 		}
 		urlStrings, err = lookupFileIdFn(context.Background(), chunk.FileId)
@@ -453,3 +507,22 @@ func eachChunk(buf []byte, eachLogEntryFn log_buffer.EachLogEntryFuncType) (proc

 	return
 }
+
+// getBufferStartFromLogFile extracts the buffer_start index from log file extended metadata
+func getBufferStartFromLogFile(logFile *filer_pb.Entry) int64 {
+	if logFile.Extended == nil {
+		return 0
+	}
+
+	// Parse buffer_start binary format
+	if startData, exists := logFile.Extended["buffer_start"]; exists {
+		if len(startData) == 8 {
+			startIndex := int64(binary.BigEndian.Uint64(startData))
+			if startIndex > 0 {
+				return startIndex
+			}
+		}
+	}
+
+	return 0
+}
--- a/weed/mq/logstore/merged_read.go
+++ b/weed/mq/logstore/merged_read.go
@@ -9,17 +9,19 @@ import (
 func GenMergedReadFunc(filerClient filer_pb.FilerClient, t topic.Topic, p topic.Partition) log_buffer.LogReadFromDiskFuncType {
 	fromParquetFn := GenParquetReadFunc(filerClient, t, p)
 	readLogDirectFn := GenLogOnDiskReadFunc(filerClient, t, p)
-	return mergeReadFuncs(fromParquetFn, readLogDirectFn)
+	// Reversed order: live logs first (recent), then Parquet files (historical)
+	// This provides better performance for real-time analytics queries
+	return mergeReadFuncs(readLogDirectFn, fromParquetFn)
 }

-func mergeReadFuncs(fromParquetFn, readLogDirectFn log_buffer.LogReadFromDiskFuncType) log_buffer.LogReadFromDiskFuncType {
-	var exhaustedParquet bool
+func mergeReadFuncs(readLogDirectFn, fromParquetFn log_buffer.LogReadFromDiskFuncType) log_buffer.LogReadFromDiskFuncType {
+	var exhaustedLiveLogs bool
 	var lastProcessedPosition log_buffer.MessagePosition
 	return func(startPosition log_buffer.MessagePosition, stopTsNs int64, eachLogEntryFn log_buffer.EachLogEntryFuncType) (lastReadPosition log_buffer.MessagePosition, isDone bool, err error) {
-		if !exhaustedParquet {
-			// glog.V(4).Infof("reading from parquet startPosition: %v\n", startPosition.UTC())
-			lastReadPosition, isDone, err = fromParquetFn(startPosition, stopTsNs, eachLogEntryFn)
-			// glog.V(4).Infof("read from parquet: %v %v %v %v\n", startPosition, lastReadPosition, isDone, err)
+		if !exhaustedLiveLogs {
+			// glog.V(4).Infof("reading from live logs startPosition: %v\n", startPosition.UTC())
+			lastReadPosition, isDone, err = readLogDirectFn(startPosition, stopTsNs, eachLogEntryFn)
+			// glog.V(4).Infof("read from live logs: %v %v %v %v\n", startPosition, lastReadPosition, isDone, err)
 			if isDone {
 				isDone = false
 			}
@@ -28,14 +30,14 @@ func mergeReadFuncs(fromParquetFn, readLogDirectFn log_buffer.LogReadFromDiskFun
 			}
 			lastProcessedPosition = lastReadPosition
 		}
-		exhaustedParquet = true
+		exhaustedLiveLogs = true

 		if startPosition.Before(lastProcessedPosition.Time) {
 			startPosition = lastProcessedPosition
 		}

-		// glog.V(4).Infof("reading from direct log startPosition: %v\n", startPosition.UTC())
-		lastReadPosition, isDone, err = readLogDirectFn(startPosition, stopTsNs, eachLogEntryFn)
+		// glog.V(4).Infof("reading from parquet startPosition: %v\n", startPosition.UTC())
+		lastReadPosition, isDone, err = fromParquetFn(startPosition, stopTsNs, eachLogEntryFn)
 		return
 	}
 }
--- a/weed/mq/logstore/read_log_from_disk.go
+++ b/weed/mq/logstore/read_log_from_disk.go
@@ -3,6 +3,10 @@ package logstore
 import (
 	"context"
 	"fmt"
+	"math"
+	"strings"
+	"time"
+
 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/mq/topic"
@@ -11,9 +15,6 @@ import (
 	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
 	"github.com/seaweedfs/seaweedfs/weed/util/log_buffer"
 	"google.golang.org/protobuf/proto"
-	"math"
-	"strings"
-	"time"
 )

 func GenLogOnDiskReadFunc(filerClient filer_pb.FilerClient, t topic.Topic, p topic.Partition) log_buffer.LogReadFromDiskFuncType {
@@ -90,7 +91,6 @@ func GenLogOnDiskReadFunc(filerClient filer_pb.FilerClient, t topic.Topic, p top
 			for _, urlString := range urlStrings {
 				// TODO optimization opportunity: reuse the buffer
 				var data []byte
-				// fmt.Printf("reading %s/%s %s\n", partitionDir, entry.Name, urlString)
 				if data, _, err = util_http.Get(urlString); err == nil {
 					processed = true
 					if processedTsNs, err = eachChunkFn(data, eachLogEntryFn, starTsNs, stopTsNs); err != nil {
--- a/weed/mq/logstore/read_parquet_to_log.go
+++ b/weed/mq/logstore/read_parquet_to_log.go
@@ -23,6 +23,34 @@ var (
 	chunkCache = chunk_cache.NewChunkCacheInMemory(256) // 256 entries, 8MB max per entry
 )

+// isControlEntry checks if a log entry is a control entry without actual data
+// Based on MQ system analysis, control entries are:
+// 1. DataMessages with populated Ctrl field (publisher close signals)
+// 2. Entries with empty keys (as filtered by subscriber)
+// 3. Entries with no data
+func isControlEntry(logEntry *filer_pb.LogEntry) bool {
+	// Skip entries with no data
+	if len(logEntry.Data) == 0 {
+		return true
+	}
+
+	// Skip entries with empty keys (same logic as subscriber)
+	if len(logEntry.Key) == 0 {
+		return true
+	}
+
+	// Check if this is a DataMessage with control field populated
+	dataMessage := &mq_pb.DataMessage{}
+	if err := proto.Unmarshal(logEntry.Data, dataMessage); err == nil {
+		// If it has a control field, it's a control message
+		if dataMessage.Ctrl != nil {
+			return true
+		}
+	}
+
+	return false
+}
+
 func GenParquetReadFunc(filerClient filer_pb.FilerClient, t topic.Topic, p topic.Partition) log_buffer.LogReadFromDiskFuncType {
 	partitionDir := topic.PartitionDir(t, p)

@@ -35,9 +63,18 @@ func GenParquetReadFunc(filerClient filer_pb.FilerClient, t topic.Topic, p topic
 		topicConf, err = t.ReadConfFile(client)
 		return err
 	}); err != nil {
-		return nil
+		// Return a no-op function for test environments or when topic config can't be read
+		return func(startPosition log_buffer.MessagePosition, stopTsNs int64, eachLogEntryFn log_buffer.EachLogEntryFuncType) (log_buffer.MessagePosition, bool, error) {
+			return startPosition, true, nil
+		}
 	}
 	recordType := topicConf.GetRecordType()
+	if recordType == nil {
+		// Return a no-op function if no schema is available
+		return func(startPosition log_buffer.MessagePosition, stopTsNs int64, eachLogEntryFn log_buffer.EachLogEntryFuncType) (log_buffer.MessagePosition, bool, error) {
+			return startPosition, true, nil
+		}
+	}
 	recordType = schema.NewRecordTypeBuilder(recordType).
 		WithField(SW_COLUMN_NAME_TS, schema.TypeInt64).
 		WithField(SW_COLUMN_NAME_KEY, schema.TypeBytes).
@@ -90,6 +127,11 @@ func GenParquetReadFunc(filerClient filer_pb.FilerClient, t topic.Topic, p topic
 					Data: data,
 				}

+				// Skip control entries without actual data
+				if isControlEntry(logEntry) {
+					continue
+				}
+
 				// fmt.Printf(" parquet entry %s ts %v\n", string(logEntry.Key), time.Unix(0, logEntry.TsNs).UTC())

 				if _, err = eachLogEntryFn(logEntry); err != nil {
@@ -108,7 +150,6 @@ func GenParquetReadFunc(filerClient filer_pb.FilerClient, t topic.Topic, p topic
 				return processedTsNs, nil
 			}
 		}
-		return
 	}

 	return func(startPosition log_buffer.MessagePosition, stopTsNs int64, eachLogEntryFn log_buffer.EachLogEntryFuncType) (lastReadPosition log_buffer.MessagePosition, isDone bool, err error) {
--- a/weed/mq/logstore/write_rows_no_panic_test.go
+++ b/weed/mq/logstore/write_rows_no_panic_test.go
@@ -0,0 +1,118 @@
+package logstore
+
+import (
+	"os"
+	"testing"
+
+	parquet "github.com/parquet-go/parquet-go"
+	"github.com/parquet-go/parquet-go/compress/zstd"
+	"github.com/seaweedfs/seaweedfs/weed/mq/schema"
+	"github.com/seaweedfs/seaweedfs/weed/pb/schema_pb"
+)
+
+// TestWriteRowsNoPanic builds a representative schema and rows and ensures WriteRows completes without panic.
+func TestWriteRowsNoPanic(t *testing.T) {
+	// Build schema similar to ecommerce.user_events
+	recordType := schema.RecordTypeBegin().
+		WithField("id", schema.TypeInt64).
+		WithField("user_id", schema.TypeInt64).
+		WithField("user_type", schema.TypeString).
+		WithField("action", schema.TypeString).
+		WithField("status", schema.TypeString).
+		WithField("amount", schema.TypeDouble).
+		WithField("timestamp", schema.TypeString).
+		WithField("metadata", schema.TypeString).
+		RecordTypeEnd()
+
+	// Add log columns
+	recordType = schema.NewRecordTypeBuilder(recordType).
+		WithField(SW_COLUMN_NAME_TS, schema.TypeInt64).
+		WithField(SW_COLUMN_NAME_KEY, schema.TypeBytes).
+		RecordTypeEnd()
+
+	ps, err := schema.ToParquetSchema("synthetic", recordType)
+	if err != nil {
+		t.Fatalf("schema: %v", err)
+	}
+	levels, err := schema.ToParquetLevels(recordType)
+	if err != nil {
+		t.Fatalf("levels: %v", err)
+	}
+
+	tmp, err := os.CreateTemp(".", "synthetic*.parquet")
+	if err != nil {
+		t.Fatalf("tmp: %v", err)
+	}
+	defer func() {
+		tmp.Close()
+		os.Remove(tmp.Name())
+	}()
+
+	w := parquet.NewWriter(tmp, ps,
+		parquet.Compression(&zstd.Codec{Level: zstd.DefaultLevel}),
+		parquet.DataPageStatistics(true),
+	)
+	defer w.Close()
+
+	rb := parquet.NewRowBuilder(ps)
+	var rows []parquet.Row
+
+	// Build a few hundred rows with various optional/missing values and nil/empty keys
+	for i := 0; i < 200; i++ {
+		rb.Reset()
+
+		rec := &schema_pb.RecordValue{Fields: map[string]*schema_pb.Value{}}
+		// Required-like fields present
+		rec.Fields["id"] = &schema_pb.Value{Kind: &schema_pb.Value_Int64Value{Int64Value: int64(1000 + i)}}
+		rec.Fields["user_id"] = &schema_pb.Value{Kind: &schema_pb.Value_Int64Value{Int64Value: int64(i)}}
+		rec.Fields["user_type"] = &schema_pb.Value{Kind: &schema_pb.Value_StringValue{StringValue: "standard"}}
+		rec.Fields["action"] = &schema_pb.Value{Kind: &schema_pb.Value_StringValue{StringValue: "click"}}
+		rec.Fields["status"] = &schema_pb.Value{Kind: &schema_pb.Value_StringValue{StringValue: "active"}}
+
+		// Optional fields vary: sometimes omitted, sometimes empty
+		if i%3 == 0 {
+			rec.Fields["amount"] = &schema_pb.Value{Kind: &schema_pb.Value_DoubleValue{DoubleValue: float64(i)}}
+		}
+		if i%4 == 0 {
+			rec.Fields["metadata"] = &schema_pb.Value{Kind: &schema_pb.Value_StringValue{StringValue: ""}}
+		}
+		if i%5 == 0 {
+			rec.Fields["timestamp"] = &schema_pb.Value{Kind: &schema_pb.Value_StringValue{StringValue: "2025-09-03T15:36:29Z"}}
+		}
+
+		// Log columns
+		rec.Fields[SW_COLUMN_NAME_TS] = &schema_pb.Value{Kind: &schema_pb.Value_Int64Value{Int64Value: int64(1756913789000000000 + i)}}
+		var keyBytes []byte
+		if i%7 == 0 {
+			keyBytes = nil // ensure nil-keys are handled
+		} else if i%7 == 1 {
+			keyBytes = []byte{} // empty
+		} else {
+			keyBytes = []byte("key-")
+		}
+		rec.Fields[SW_COLUMN_NAME_KEY] = &schema_pb.Value{Kind: &schema_pb.Value_BytesValue{BytesValue: keyBytes}}
+
+		if err := schema.AddRecordValue(rb, recordType, levels, rec); err != nil {
+			t.Fatalf("add record: %v", err)
+		}
+		rows = append(rows, rb.Row())
+	}
+
+	deferredPanicked := false
+	defer func() {
+		if r := recover(); r != nil {
+			deferredPanicked = true
+			t.Fatalf("unexpected panic: %v", r)
+		}
+	}()
+
+	if _, err := w.WriteRows(rows); err != nil {
+		t.Fatalf("WriteRows: %v", err)
+	}
+	if err := w.Close(); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+	if deferredPanicked {
+		t.Fatal("panicked")
+	}
+}
--- a/weed/mq/schema/schema_builder.go
+++ b/weed/mq/schema/schema_builder.go
@@ -1,11 +1,13 @@
 package schema

 import (
-	"github.com/seaweedfs/seaweedfs/weed/pb/schema_pb"
 	"sort"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/schema_pb"
 )

 var (
+	// Basic scalar types
 	TypeBoolean = &schema_pb.Type{Kind: &schema_pb.Type_ScalarType{schema_pb.ScalarType_BOOL}}
 	TypeInt32   = &schema_pb.Type{Kind: &schema_pb.Type_ScalarType{schema_pb.ScalarType_INT32}}
 	TypeInt64   = &schema_pb.Type{Kind: &schema_pb.Type_ScalarType{schema_pb.ScalarType_INT64}}
@@ -13,6 +15,12 @@ var (
 	TypeDouble  = &schema_pb.Type{Kind: &schema_pb.Type_ScalarType{schema_pb.ScalarType_DOUBLE}}
 	TypeBytes   = &schema_pb.Type{Kind: &schema_pb.Type_ScalarType{schema_pb.ScalarType_BYTES}}
 	TypeString  = &schema_pb.Type{Kind: &schema_pb.Type_ScalarType{schema_pb.ScalarType_STRING}}
+
+	// Parquet logical types
+	TypeTimestamp = &schema_pb.Type{Kind: &schema_pb.Type_ScalarType{schema_pb.ScalarType_TIMESTAMP}}
+	TypeDate      = &schema_pb.Type{Kind: &schema_pb.Type_ScalarType{schema_pb.ScalarType_DATE}}
+	TypeDecimal   = &schema_pb.Type{Kind: &schema_pb.Type_ScalarType{schema_pb.ScalarType_DECIMAL}}
+	TypeTime      = &schema_pb.Type{Kind: &schema_pb.Type_ScalarType{schema_pb.ScalarType_TIME}}
 )

 type RecordTypeBuilder struct {
--- a/weed/mq/schema/struct_to_schema.go
+++ b/weed/mq/schema/struct_to_schema.go
@@ -1,8 +1,9 @@
 package schema

 import (
-	"github.com/seaweedfs/seaweedfs/weed/pb/schema_pb"
 	"reflect"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/schema_pb"
 )

 func StructToSchema(instance any) *schema_pb.RecordType {
--- a/weed/mq/schema/to_parquet_schema.go
+++ b/weed/mq/schema/to_parquet_schema.go
@@ -2,6 +2,7 @@ package schema

 import (
 	"fmt"
+
 	parquet "github.com/parquet-go/parquet-go"
 	"github.com/seaweedfs/seaweedfs/weed/pb/schema_pb"
 )
@@ -18,20 +19,8 @@ func ToParquetSchema(topicName string, recordType *schema_pb.RecordType) (*parqu
 }

 func toParquetFieldType(fieldType *schema_pb.Type) (dataType parquet.Node, err error) {
-	switch fieldType.Kind.(type) {
-	case *schema_pb.Type_ScalarType:
-		dataType, err = toParquetFieldTypeScalar(fieldType.GetScalarType())
-		dataType = parquet.Optional(dataType)
-	case *schema_pb.Type_RecordType:
-		dataType, err = toParquetFieldTypeRecord(fieldType.GetRecordType())
-		dataType = parquet.Optional(dataType)
-	case *schema_pb.Type_ListType:
-		dataType, err = toParquetFieldTypeList(fieldType.GetListType())
-	default:
-		return nil, fmt.Errorf("unknown field type: %T", fieldType.Kind)
-	}
-
-	return dataType, err
+	// This is the old function - now defaults to Optional for backward compatibility
+	return toParquetFieldTypeWithRequirement(fieldType, false)
 }

 func toParquetFieldTypeList(listType *schema_pb.ListType) (parquet.Node, error) {
@@ -58,6 +47,22 @@ func toParquetFieldTypeScalar(scalarType schema_pb.ScalarType) (parquet.Node, er
 		return parquet.Leaf(parquet.ByteArrayType), nil
 	case schema_pb.ScalarType_STRING:
 		return parquet.Leaf(parquet.ByteArrayType), nil
+	// Parquet logical types - map to their physical storage types
+	case schema_pb.ScalarType_TIMESTAMP:
+		// Stored as INT64 (microseconds since Unix epoch)
+		return parquet.Leaf(parquet.Int64Type), nil
+	case schema_pb.ScalarType_DATE:
+		// Stored as INT32 (days since Unix epoch)
+		return parquet.Leaf(parquet.Int32Type), nil
+	case schema_pb.ScalarType_DECIMAL:
+		// Use maximum precision/scale to accommodate any decimal value
+		// Per Parquet spec: precision ≤9→INT32, ≤18→INT64, >18→FixedLenByteArray
+		// Using precision=38 (max for most systems), scale=18 for flexibility
+		// Individual values can have smaller precision/scale, but schema supports maximum
+		return parquet.Decimal(18, 38, parquet.FixedLenByteArrayType(16)), nil
+	case schema_pb.ScalarType_TIME:
+		// Stored as INT64 (microseconds since midnight)
+		return parquet.Leaf(parquet.Int64Type), nil
 	default:
 		return nil, fmt.Errorf("unknown scalar type: %v", scalarType)
 	}
@@ -65,7 +70,7 @@ func toParquetFieldTypeScalar(scalarType schema_pb.ScalarType) (parquet.Node, er
 func toParquetFieldTypeRecord(recordType *schema_pb.RecordType) (parquet.Node, error) {
 	recordNode := parquet.Group{}
 	for _, field := range recordType.Fields {
-		parquetFieldType, err := toParquetFieldType(field.Type)
+		parquetFieldType, err := toParquetFieldTypeWithRequirement(field.Type, field.IsRequired)
 		if err != nil {
 			return nil, err
 		}
@@ -73,3 +78,40 @@ func toParquetFieldTypeRecord(recordType *schema_pb.RecordType) (parquet.Node, e
 	}
 	return recordNode, nil
 }
+
+// toParquetFieldTypeWithRequirement creates parquet field type respecting required/optional constraints
+func toParquetFieldTypeWithRequirement(fieldType *schema_pb.Type, isRequired bool) (dataType parquet.Node, err error) {
+	switch fieldType.Kind.(type) {
+	case *schema_pb.Type_ScalarType:
+		dataType, err = toParquetFieldTypeScalar(fieldType.GetScalarType())
+		if err != nil {
+			return nil, err
+		}
+		if isRequired {
+			// Required fields are NOT wrapped in Optional
+			return dataType, nil
+		} else {
+			// Optional fields are wrapped in Optional
+			return parquet.Optional(dataType), nil
+		}
+	case *schema_pb.Type_RecordType:
+		dataType, err = toParquetFieldTypeRecord(fieldType.GetRecordType())
+		if err != nil {
+			return nil, err
+		}
+		if isRequired {
+			return dataType, nil
+		} else {
+			return parquet.Optional(dataType), nil
+		}
+	case *schema_pb.Type_ListType:
+		dataType, err = toParquetFieldTypeList(fieldType.GetListType())
+		if err != nil {
+			return nil, err
+		}
+		// Lists are typically optional by nature
+		return dataType, nil
+	default:
+		return nil, fmt.Errorf("unknown field type: %T", fieldType.Kind)
+	}
+}
--- a/weed/mq/schema/to_parquet_value.go
+++ b/weed/mq/schema/to_parquet_value.go
@@ -2,6 +2,8 @@ package schema

 import (
 	"fmt"
+	"strconv"
+
 	parquet "github.com/parquet-go/parquet-go"
 	"github.com/seaweedfs/seaweedfs/weed/pb/schema_pb"
 )
@@ -9,16 +11,32 @@ import (
 func rowBuilderVisit(rowBuilder *parquet.RowBuilder, fieldType *schema_pb.Type, levels *ParquetLevels, fieldValue *schema_pb.Value) (err error) {
 	switch fieldType.Kind.(type) {
 	case *schema_pb.Type_ScalarType:
+		// If value is missing, write NULL at the correct column to keep rows aligned
+		if fieldValue == nil || fieldValue.Kind == nil {
+			rowBuilder.Add(levels.startColumnIndex, parquet.NullValue())
+			return nil
+		}
 		var parquetValue parquet.Value
-		parquetValue, err = toParquetValue(fieldValue)
+		parquetValue, err = toParquetValueForType(fieldType, fieldValue)
 		if err != nil {
 			return
 		}
+
+		// Safety check: prevent nil byte arrays from reaching parquet library
+		if parquetValue.Kind() == parquet.ByteArray {
+			byteData := parquetValue.ByteArray()
+			if byteData == nil {
+				parquetValue = parquet.ByteArrayValue([]byte{})
+			}
+		}
+
 		rowBuilder.Add(levels.startColumnIndex, parquetValue)
-		// fmt.Printf("rowBuilder.Add %d %v\n", columnIndex, parquetValue)
 	case *schema_pb.Type_ListType:
+		// Advance to list position even if value is missing
 		rowBuilder.Next(levels.startColumnIndex)
-		// fmt.Printf("rowBuilder.Next %d\n", columnIndex)
+		if fieldValue == nil || fieldValue.GetListValue() == nil {
+			return nil
+		}

 		elementType := fieldType.GetListType().ElementType
 		for _, value := range fieldValue.GetListValue().Values {
@@ -54,13 +72,17 @@ func doVisitValue(fieldType *schema_pb.Type, levels *ParquetLevels, fieldValue *
 		return visitor(fieldType, levels, fieldValue)
 	case *schema_pb.Type_RecordType:
 		for _, field := range fieldType.GetRecordType().Fields {
-			fieldValue, found := fieldValue.GetRecordValue().Fields[field.Name]
-			if !found {
-				// TODO check this if no such field found
-				continue
+			var fv *schema_pb.Value
+			if fieldValue != nil && fieldValue.GetRecordValue() != nil {
+				var found bool
+				fv, found = fieldValue.GetRecordValue().Fields[field.Name]
+				if !found {
+					// pass nil so visitor can emit NULL for alignment
+					fv = nil
+				}
 			}
 			fieldLevels := levels.levels[field.Name]
-			err = doVisitValue(field.Type, fieldLevels, fieldValue, visitor)
+			err = doVisitValue(field.Type, fieldLevels, fv, visitor)
 			if err != nil {
 				return
 			}
@@ -71,6 +93,11 @@ func doVisitValue(fieldType *schema_pb.Type, levels *ParquetLevels, fieldValue *
 }

 func toParquetValue(value *schema_pb.Value) (parquet.Value, error) {
+	// Safety check for nil value
+	if value == nil || value.Kind == nil {
+		return parquet.NullValue(), fmt.Errorf("nil value or nil value kind")
+	}
+
 	switch value.Kind.(type) {
 	case *schema_pb.Value_BoolValue:
 		return parquet.BooleanValue(value.GetBoolValue()), nil
@@ -83,10 +110,237 @@ func toParquetValue(value *schema_pb.Value) (parquet.Value, error) {
 	case *schema_pb.Value_DoubleValue:
 		return parquet.DoubleValue(value.GetDoubleValue()), nil
 	case *schema_pb.Value_BytesValue:
-		return parquet.ByteArrayValue(value.GetBytesValue()), nil
+		// Handle nil byte slices to prevent growslice panic in parquet-go
+		byteData := value.GetBytesValue()
+		if byteData == nil {
+			byteData = []byte{} // Use empty slice instead of nil
+		}
+		return parquet.ByteArrayValue(byteData), nil
 	case *schema_pb.Value_StringValue:
-		return parquet.ByteArrayValue([]byte(value.GetStringValue())), nil
+		// Convert string to bytes, ensuring we never pass nil
+		stringData := value.GetStringValue()
+		return parquet.ByteArrayValue([]byte(stringData)), nil
+	// Parquet logical types with safe conversion (preventing commit 7a4aeec60 panic)
+	case *schema_pb.Value_TimestampValue:
+		timestampValue := value.GetTimestampValue()
+		if timestampValue == nil {
+			return parquet.NullValue(), nil
+		}
+		return parquet.Int64Value(timestampValue.TimestampMicros), nil
+	case *schema_pb.Value_DateValue:
+		dateValue := value.GetDateValue()
+		if dateValue == nil {
+			return parquet.NullValue(), nil
+		}
+		return parquet.Int32Value(dateValue.DaysSinceEpoch), nil
+	case *schema_pb.Value_DecimalValue:
+		decimalValue := value.GetDecimalValue()
+		if decimalValue == nil || decimalValue.Value == nil || len(decimalValue.Value) == 0 {
+			return parquet.NullValue(), nil
+		}
+
+		// Validate input data - reject unreasonably large values instead of corrupting data
+		if len(decimalValue.Value) > 64 {
+			// Reject extremely large decimal values (>512 bits) as likely corrupted data
+			// Better to fail fast than silently corrupt financial/scientific data
+			return parquet.NullValue(), fmt.Errorf("decimal value too large: %d bytes (max 64)", len(decimalValue.Value))
+		}
+
+		// Convert to FixedLenByteArray to match schema (DECIMAL with FixedLenByteArray physical type)
+		// This accommodates any precision up to 38 digits (16 bytes = 128 bits)
+
+		// Pad or truncate to exactly 16 bytes for FixedLenByteArray
+		fixedBytes := make([]byte, 16)
+		if len(decimalValue.Value) <= 16 {
+			// Right-align the value (big-endian)
+			copy(fixedBytes[16-len(decimalValue.Value):], decimalValue.Value)
+		} else {
+			// Truncate if too large, taking the least significant bytes
+			copy(fixedBytes, decimalValue.Value[len(decimalValue.Value)-16:])
+		}
+
+		return parquet.FixedLenByteArrayValue(fixedBytes), nil
+	case *schema_pb.Value_TimeValue:
+		timeValue := value.GetTimeValue()
+		if timeValue == nil {
+			return parquet.NullValue(), nil
+		}
+		return parquet.Int64Value(timeValue.TimeMicros), nil
 	default:
 		return parquet.NullValue(), fmt.Errorf("unknown value type: %T", value.Kind)
 	}
 }
+
+// toParquetValueForType coerces a schema_pb.Value into a parquet.Value that matches the declared field type.
+func toParquetValueForType(fieldType *schema_pb.Type, value *schema_pb.Value) (parquet.Value, error) {
+	switch t := fieldType.Kind.(type) {
+	case *schema_pb.Type_ScalarType:
+		switch t.ScalarType {
+		case schema_pb.ScalarType_BOOL:
+			switch v := value.Kind.(type) {
+			case *schema_pb.Value_BoolValue:
+				return parquet.BooleanValue(v.BoolValue), nil
+			case *schema_pb.Value_StringValue:
+				if b, err := strconv.ParseBool(v.StringValue); err == nil {
+					return parquet.BooleanValue(b), nil
+				}
+				return parquet.BooleanValue(false), nil
+			default:
+				return parquet.BooleanValue(false), nil
+			}
+
+		case schema_pb.ScalarType_INT32:
+			switch v := value.Kind.(type) {
+			case *schema_pb.Value_Int32Value:
+				return parquet.Int32Value(v.Int32Value), nil
+			case *schema_pb.Value_Int64Value:
+				return parquet.Int32Value(int32(v.Int64Value)), nil
+			case *schema_pb.Value_DoubleValue:
+				return parquet.Int32Value(int32(v.DoubleValue)), nil
+			case *schema_pb.Value_StringValue:
+				if i, err := strconv.ParseInt(v.StringValue, 10, 32); err == nil {
+					return parquet.Int32Value(int32(i)), nil
+				}
+				return parquet.Int32Value(0), nil
+			default:
+				return parquet.Int32Value(0), nil
+			}
+
+		case schema_pb.ScalarType_INT64:
+			switch v := value.Kind.(type) {
+			case *schema_pb.Value_Int64Value:
+				return parquet.Int64Value(v.Int64Value), nil
+			case *schema_pb.Value_Int32Value:
+				return parquet.Int64Value(int64(v.Int32Value)), nil
+			case *schema_pb.Value_DoubleValue:
+				return parquet.Int64Value(int64(v.DoubleValue)), nil
+			case *schema_pb.Value_StringValue:
+				if i, err := strconv.ParseInt(v.StringValue, 10, 64); err == nil {
+					return parquet.Int64Value(i), nil
+				}
+				return parquet.Int64Value(0), nil
+			default:
+				return parquet.Int64Value(0), nil
+			}
+
+		case schema_pb.ScalarType_FLOAT:
+			switch v := value.Kind.(type) {
+			case *schema_pb.Value_FloatValue:
+				return parquet.FloatValue(v.FloatValue), nil
+			case *schema_pb.Value_DoubleValue:
+				return parquet.FloatValue(float32(v.DoubleValue)), nil
+			case *schema_pb.Value_Int64Value:
+				return parquet.FloatValue(float32(v.Int64Value)), nil
+			case *schema_pb.Value_StringValue:
+				if f, err := strconv.ParseFloat(v.StringValue, 32); err == nil {
+					return parquet.FloatValue(float32(f)), nil
+				}
+				return parquet.FloatValue(0), nil
+			default:
+				return parquet.FloatValue(0), nil
+			}
+
+		case schema_pb.ScalarType_DOUBLE:
+			switch v := value.Kind.(type) {
+			case *schema_pb.Value_DoubleValue:
+				return parquet.DoubleValue(v.DoubleValue), nil
+			case *schema_pb.Value_Int64Value:
+				return parquet.DoubleValue(float64(v.Int64Value)), nil
+			case *schema_pb.Value_Int32Value:
+				return parquet.DoubleValue(float64(v.Int32Value)), nil
+			case *schema_pb.Value_StringValue:
+				if f, err := strconv.ParseFloat(v.StringValue, 64); err == nil {
+					return parquet.DoubleValue(f), nil
+				}
+				return parquet.DoubleValue(0), nil
+			default:
+				return parquet.DoubleValue(0), nil
+			}
+
+		case schema_pb.ScalarType_BYTES:
+			switch v := value.Kind.(type) {
+			case *schema_pb.Value_BytesValue:
+				b := v.BytesValue
+				if b == nil {
+					b = []byte{}
+				}
+				return parquet.ByteArrayValue(b), nil
+			case *schema_pb.Value_StringValue:
+				return parquet.ByteArrayValue([]byte(v.StringValue)), nil
+			case *schema_pb.Value_Int64Value:
+				return parquet.ByteArrayValue([]byte(strconv.FormatInt(v.Int64Value, 10))), nil
+			case *schema_pb.Value_Int32Value:
+				return parquet.ByteArrayValue([]byte(strconv.FormatInt(int64(v.Int32Value), 10))), nil
+			case *schema_pb.Value_DoubleValue:
+				return parquet.ByteArrayValue([]byte(strconv.FormatFloat(v.DoubleValue, 'f', -1, 64))), nil
+			case *schema_pb.Value_FloatValue:
+				return parquet.ByteArrayValue([]byte(strconv.FormatFloat(float64(v.FloatValue), 'f', -1, 32))), nil
+			case *schema_pb.Value_BoolValue:
+				if v.BoolValue {
+					return parquet.ByteArrayValue([]byte("true")), nil
+				}
+				return parquet.ByteArrayValue([]byte("false")), nil
+			default:
+				return parquet.ByteArrayValue([]byte{}), nil
+			}
+
+		case schema_pb.ScalarType_STRING:
+			// Same as bytes but semantically string
+			switch v := value.Kind.(type) {
+			case *schema_pb.Value_StringValue:
+				return parquet.ByteArrayValue([]byte(v.StringValue)), nil
+			default:
+				// Fallback through bytes coercion
+				b, _ := toParquetValueForType(&schema_pb.Type{Kind: &schema_pb.Type_ScalarType{ScalarType: schema_pb.ScalarType_BYTES}}, value)
+				return b, nil
+			}
+
+		case schema_pb.ScalarType_TIMESTAMP:
+			switch v := value.Kind.(type) {
+			case *schema_pb.Value_Int64Value:
+				return parquet.Int64Value(v.Int64Value), nil
+			case *schema_pb.Value_StringValue:
+				if i, err := strconv.ParseInt(v.StringValue, 10, 64); err == nil {
+					return parquet.Int64Value(i), nil
+				}
+				return parquet.Int64Value(0), nil
+			default:
+				return parquet.Int64Value(0), nil
+			}
+
+		case schema_pb.ScalarType_DATE:
+			switch v := value.Kind.(type) {
+			case *schema_pb.Value_Int32Value:
+				return parquet.Int32Value(v.Int32Value), nil
+			case *schema_pb.Value_Int64Value:
+				return parquet.Int32Value(int32(v.Int64Value)), nil
+			case *schema_pb.Value_StringValue:
+				if i, err := strconv.ParseInt(v.StringValue, 10, 32); err == nil {
+					return parquet.Int32Value(int32(i)), nil
+				}
+				return parquet.Int32Value(0), nil
+			default:
+				return parquet.Int32Value(0), nil
+			}
+
+		case schema_pb.ScalarType_DECIMAL:
+			// Reuse existing conversion path (FixedLenByteArray 16)
+			return toParquetValue(value)
+
+		case schema_pb.ScalarType_TIME:
+			switch v := value.Kind.(type) {
+			case *schema_pb.Value_Int64Value:
+				return parquet.Int64Value(v.Int64Value), nil
+			case *schema_pb.Value_StringValue:
+				if i, err := strconv.ParseInt(v.StringValue, 10, 64); err == nil {
+					return parquet.Int64Value(i), nil
+				}
+				return parquet.Int64Value(0), nil
+			default:
+				return parquet.Int64Value(0), nil
+			}
+		}
+	}
+	// Fallback to generic conversion
+	return toParquetValue(value)
+}
--- a/weed/mq/schema/to_parquet_value_test.go
+++ b/weed/mq/schema/to_parquet_value_test.go
@@ -0,0 +1,666 @@
+package schema
+
+import (
+	"math/big"
+	"testing"
+	"time"
+
+	"github.com/parquet-go/parquet-go"
+	"github.com/seaweedfs/seaweedfs/weed/pb/schema_pb"
+)
+
+func TestToParquetValue_BasicTypes(t *testing.T) {
+	tests := []struct {
+		name     string
+		value    *schema_pb.Value
+		expected parquet.Value
+		wantErr  bool
+	}{
+		{
+			name: "BoolValue true",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_BoolValue{BoolValue: true},
+			},
+			expected: parquet.BooleanValue(true),
+		},
+		{
+			name: "Int32Value",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_Int32Value{Int32Value: 42},
+			},
+			expected: parquet.Int32Value(42),
+		},
+		{
+			name: "Int64Value",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_Int64Value{Int64Value: 12345678901234},
+			},
+			expected: parquet.Int64Value(12345678901234),
+		},
+		{
+			name: "FloatValue",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_FloatValue{FloatValue: 3.14159},
+			},
+			expected: parquet.FloatValue(3.14159),
+		},
+		{
+			name: "DoubleValue",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_DoubleValue{DoubleValue: 2.718281828},
+			},
+			expected: parquet.DoubleValue(2.718281828),
+		},
+		{
+			name: "BytesValue",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_BytesValue{BytesValue: []byte("hello world")},
+			},
+			expected: parquet.ByteArrayValue([]byte("hello world")),
+		},
+		{
+			name: "BytesValue empty",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_BytesValue{BytesValue: []byte{}},
+			},
+			expected: parquet.ByteArrayValue([]byte{}),
+		},
+		{
+			name: "StringValue",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_StringValue{StringValue: "test string"},
+			},
+			expected: parquet.ByteArrayValue([]byte("test string")),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result, err := toParquetValue(tt.value)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("toParquetValue() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !parquetValuesEqual(result, tt.expected) {
+				t.Errorf("toParquetValue() = %v, want %v", result, tt.expected)
+			}
+		})
+	}
+}
+
+func TestToParquetValue_TimestampValue(t *testing.T) {
+	tests := []struct {
+		name     string
+		value    *schema_pb.Value
+		expected parquet.Value
+		wantErr  bool
+	}{
+		{
+			name: "Valid TimestampValue UTC",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_TimestampValue{
+					TimestampValue: &schema_pb.TimestampValue{
+						TimestampMicros: 1704067200000000, // 2024-01-01 00:00:00 UTC in microseconds
+						IsUtc:           true,
+					},
+				},
+			},
+			expected: parquet.Int64Value(1704067200000000),
+		},
+		{
+			name: "Valid TimestampValue local",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_TimestampValue{
+					TimestampValue: &schema_pb.TimestampValue{
+						TimestampMicros: 1704067200000000,
+						IsUtc:           false,
+					},
+				},
+			},
+			expected: parquet.Int64Value(1704067200000000),
+		},
+		{
+			name: "TimestampValue zero",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_TimestampValue{
+					TimestampValue: &schema_pb.TimestampValue{
+						TimestampMicros: 0,
+						IsUtc:           true,
+					},
+				},
+			},
+			expected: parquet.Int64Value(0),
+		},
+		{
+			name: "TimestampValue negative (before epoch)",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_TimestampValue{
+					TimestampValue: &schema_pb.TimestampValue{
+						TimestampMicros: -1000000, // 1 second before epoch
+						IsUtc:           true,
+					},
+				},
+			},
+			expected: parquet.Int64Value(-1000000),
+		},
+		{
+			name: "TimestampValue nil pointer",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_TimestampValue{
+					TimestampValue: nil,
+				},
+			},
+			expected: parquet.NullValue(),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result, err := toParquetValue(tt.value)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("toParquetValue() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !parquetValuesEqual(result, tt.expected) {
+				t.Errorf("toParquetValue() = %v, want %v", result, tt.expected)
+			}
+		})
+	}
+}
+
+func TestToParquetValue_DateValue(t *testing.T) {
+	tests := []struct {
+		name     string
+		value    *schema_pb.Value
+		expected parquet.Value
+		wantErr  bool
+	}{
+		{
+			name: "Valid DateValue (2024-01-01)",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_DateValue{
+					DateValue: &schema_pb.DateValue{
+						DaysSinceEpoch: 19723, // 2024-01-01 = 19723 days since epoch
+					},
+				},
+			},
+			expected: parquet.Int32Value(19723),
+		},
+		{
+			name: "DateValue epoch (1970-01-01)",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_DateValue{
+					DateValue: &schema_pb.DateValue{
+						DaysSinceEpoch: 0,
+					},
+				},
+			},
+			expected: parquet.Int32Value(0),
+		},
+		{
+			name: "DateValue before epoch",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_DateValue{
+					DateValue: &schema_pb.DateValue{
+						DaysSinceEpoch: -365, // 1969-01-01
+					},
+				},
+			},
+			expected: parquet.Int32Value(-365),
+		},
+		{
+			name: "DateValue nil pointer",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_DateValue{
+					DateValue: nil,
+				},
+			},
+			expected: parquet.NullValue(),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result, err := toParquetValue(tt.value)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("toParquetValue() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !parquetValuesEqual(result, tt.expected) {
+				t.Errorf("toParquetValue() = %v, want %v", result, tt.expected)
+			}
+		})
+	}
+}
+
+func TestToParquetValue_DecimalValue(t *testing.T) {
+	tests := []struct {
+		name     string
+		value    *schema_pb.Value
+		expected parquet.Value
+		wantErr  bool
+	}{
+		{
+			name: "Small Decimal (precision <= 9) - positive",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_DecimalValue{
+					DecimalValue: &schema_pb.DecimalValue{
+						Value:     encodeBigIntToBytes(big.NewInt(12345)), // 123.45 with scale 2
+						Precision: 5,
+						Scale:     2,
+					},
+				},
+			},
+			expected: createFixedLenByteArray(encodeBigIntToBytes(big.NewInt(12345))), // FixedLenByteArray conversion
+		},
+		{
+			name: "Small Decimal (precision <= 9) - negative",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_DecimalValue{
+					DecimalValue: &schema_pb.DecimalValue{
+						Value:     encodeBigIntToBytes(big.NewInt(-12345)),
+						Precision: 5,
+						Scale:     2,
+					},
+				},
+			},
+			expected: createFixedLenByteArray(encodeBigIntToBytes(big.NewInt(-12345))), // FixedLenByteArray conversion
+		},
+		{
+			name: "Medium Decimal (9 < precision <= 18)",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_DecimalValue{
+					DecimalValue: &schema_pb.DecimalValue{
+						Value:     encodeBigIntToBytes(big.NewInt(123456789012345)),
+						Precision: 15,
+						Scale:     2,
+					},
+				},
+			},
+			expected: createFixedLenByteArray(encodeBigIntToBytes(big.NewInt(123456789012345))), // FixedLenByteArray conversion
+		},
+		{
+			name: "Large Decimal (precision > 18)",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_DecimalValue{
+					DecimalValue: &schema_pb.DecimalValue{
+						Value:     []byte{0x01, 0x23, 0x45, 0x67, 0x89, 0xAB, 0xCD, 0xEF}, // Large number as bytes
+						Precision: 25,
+						Scale:     5,
+					},
+				},
+			},
+			expected: createFixedLenByteArray([]byte{0x01, 0x23, 0x45, 0x67, 0x89, 0xAB, 0xCD, 0xEF}), // FixedLenByteArray conversion
+		},
+		{
+			name: "Decimal with zero precision",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_DecimalValue{
+					DecimalValue: &schema_pb.DecimalValue{
+						Value:     encodeBigIntToBytes(big.NewInt(0)),
+						Precision: 0,
+						Scale:     0,
+					},
+				},
+			},
+			expected: createFixedLenByteArray(encodeBigIntToBytes(big.NewInt(0))), // Zero as FixedLenByteArray
+		},
+		{
+			name: "Decimal nil pointer",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_DecimalValue{
+					DecimalValue: nil,
+				},
+			},
+			expected: parquet.NullValue(),
+		},
+		{
+			name: "Decimal with nil Value bytes",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_DecimalValue{
+					DecimalValue: &schema_pb.DecimalValue{
+						Value:     nil, // This was the original panic cause
+						Precision: 5,
+						Scale:     2,
+					},
+				},
+			},
+			expected: parquet.NullValue(),
+		},
+		{
+			name: "Decimal with empty Value bytes",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_DecimalValue{
+					DecimalValue: &schema_pb.DecimalValue{
+						Value:     []byte{}, // Empty slice
+						Precision: 5,
+						Scale:     2,
+					},
+				},
+			},
+			expected: parquet.NullValue(), // Returns null for empty bytes
+		},
+		{
+			name: "Decimal out of int32 range (stored as binary)",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_DecimalValue{
+					DecimalValue: &schema_pb.DecimalValue{
+						Value:     encodeBigIntToBytes(big.NewInt(999999999999)), // Too large for int32
+						Precision: 5,                                             // But precision says int32
+						Scale:     0,
+					},
+				},
+			},
+			expected: createFixedLenByteArray(encodeBigIntToBytes(big.NewInt(999999999999))), // FixedLenByteArray
+		},
+		{
+			name: "Decimal out of int64 range (stored as binary)",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_DecimalValue{
+					DecimalValue: &schema_pb.DecimalValue{
+						Value: func() []byte {
+							// Create a number larger than int64 max
+							bigNum := new(big.Int)
+							bigNum.SetString("99999999999999999999999999999", 10)
+							return encodeBigIntToBytes(bigNum)
+						}(),
+						Precision: 15, // Says int64 but value is too large
+						Scale:     0,
+					},
+				},
+			},
+			expected: createFixedLenByteArray(func() []byte {
+				bigNum := new(big.Int)
+				bigNum.SetString("99999999999999999999999999999", 10)
+				return encodeBigIntToBytes(bigNum)
+			}()), // Large number as FixedLenByteArray (truncated to 16 bytes)
+		},
+		{
+			name: "Decimal extremely large value (should be rejected)",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_DecimalValue{
+					DecimalValue: &schema_pb.DecimalValue{
+						Value:     make([]byte, 100), // 100 bytes > 64 byte limit
+						Precision: 100,
+						Scale:     0,
+					},
+				},
+			},
+			expected: parquet.NullValue(),
+			wantErr:  true, // Should return error instead of corrupting data
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result, err := toParquetValue(tt.value)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("toParquetValue() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !parquetValuesEqual(result, tt.expected) {
+				t.Errorf("toParquetValue() = %v, want %v", result, tt.expected)
+			}
+		})
+	}
+}
+
+func TestToParquetValue_TimeValue(t *testing.T) {
+	tests := []struct {
+		name     string
+		value    *schema_pb.Value
+		expected parquet.Value
+		wantErr  bool
+	}{
+		{
+			name: "Valid TimeValue (12:34:56.789)",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_TimeValue{
+					TimeValue: &schema_pb.TimeValue{
+						TimeMicros: 45296789000, // 12:34:56.789 in microseconds since midnight
+					},
+				},
+			},
+			expected: parquet.Int64Value(45296789000),
+		},
+		{
+			name: "TimeValue midnight",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_TimeValue{
+					TimeValue: &schema_pb.TimeValue{
+						TimeMicros: 0,
+					},
+				},
+			},
+			expected: parquet.Int64Value(0),
+		},
+		{
+			name: "TimeValue end of day (23:59:59.999999)",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_TimeValue{
+					TimeValue: &schema_pb.TimeValue{
+						TimeMicros: 86399999999, // 23:59:59.999999
+					},
+				},
+			},
+			expected: parquet.Int64Value(86399999999),
+		},
+		{
+			name: "TimeValue nil pointer",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_TimeValue{
+					TimeValue: nil,
+				},
+			},
+			expected: parquet.NullValue(),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result, err := toParquetValue(tt.value)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("toParquetValue() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !parquetValuesEqual(result, tt.expected) {
+				t.Errorf("toParquetValue() = %v, want %v", result, tt.expected)
+			}
+		})
+	}
+}
+
+func TestToParquetValue_EdgeCases(t *testing.T) {
+	tests := []struct {
+		name     string
+		value    *schema_pb.Value
+		expected parquet.Value
+		wantErr  bool
+	}{
+		{
+			name: "Nil value",
+			value: &schema_pb.Value{
+				Kind: nil,
+			},
+			wantErr: true,
+		},
+		{
+			name:    "Completely nil value",
+			value:   nil,
+			wantErr: true,
+		},
+		{
+			name: "BytesValue with nil slice",
+			value: &schema_pb.Value{
+				Kind: &schema_pb.Value_BytesValue{BytesValue: nil},
+			},
+			expected: parquet.ByteArrayValue([]byte{}), // Should convert nil to empty slice
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result, err := toParquetValue(tt.value)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("toParquetValue() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !tt.wantErr && !parquetValuesEqual(result, tt.expected) {
+				t.Errorf("toParquetValue() = %v, want %v", result, tt.expected)
+			}
+		})
+	}
+}
+
+// Helper function to encode a big.Int to bytes using two's complement representation
+func encodeBigIntToBytes(n *big.Int) []byte {
+	if n.Sign() == 0 {
+		return []byte{0}
+	}
+
+	// For positive numbers, just use Bytes()
+	if n.Sign() > 0 {
+		return n.Bytes()
+	}
+
+	// For negative numbers, we need two's complement representation
+	bitLen := n.BitLen()
+	if bitLen%8 != 0 {
+		bitLen += 8 - (bitLen % 8) // Round up to byte boundary
+	}
+	byteLen := bitLen / 8
+	if byteLen == 0 {
+		byteLen = 1
+	}
+
+	// Calculate 2^(byteLen*8)
+	modulus := new(big.Int).Lsh(big.NewInt(1), uint(byteLen*8))
+
+	// Convert negative to positive representation: n + 2^(byteLen*8)
+	positive := new(big.Int).Add(n, modulus)
+
+	bytes := positive.Bytes()
+
+	// Pad with leading zeros if needed
+	if len(bytes) < byteLen {
+		padded := make([]byte, byteLen)
+		copy(padded[byteLen-len(bytes):], bytes)
+		return padded
+	}
+
+	return bytes
+}
+
+// Helper function to create a FixedLenByteArray(16) matching our conversion logic
+func createFixedLenByteArray(inputBytes []byte) parquet.Value {
+	fixedBytes := make([]byte, 16)
+	if len(inputBytes) <= 16 {
+		// Right-align the value (big-endian) - same as our conversion logic
+		copy(fixedBytes[16-len(inputBytes):], inputBytes)
+	} else {
+		// Truncate if too large, taking the least significant bytes
+		copy(fixedBytes, inputBytes[len(inputBytes)-16:])
+	}
+	return parquet.FixedLenByteArrayValue(fixedBytes)
+}
+
+// Helper function to compare parquet values
+func parquetValuesEqual(a, b parquet.Value) bool {
+	// Handle both being null
+	if a.IsNull() && b.IsNull() {
+		return true
+	}
+	if a.IsNull() != b.IsNull() {
+		return false
+	}
+
+	// Compare kind first
+	if a.Kind() != b.Kind() {
+		return false
+	}
+
+	// Compare based on type
+	switch a.Kind() {
+	case parquet.Boolean:
+		return a.Boolean() == b.Boolean()
+	case parquet.Int32:
+		return a.Int32() == b.Int32()
+	case parquet.Int64:
+		return a.Int64() == b.Int64()
+	case parquet.Float:
+		return a.Float() == b.Float()
+	case parquet.Double:
+		return a.Double() == b.Double()
+	case parquet.ByteArray:
+		aBytes := a.ByteArray()
+		bBytes := b.ByteArray()
+		if len(aBytes) != len(bBytes) {
+			return false
+		}
+		for i, v := range aBytes {
+			if v != bBytes[i] {
+				return false
+			}
+		}
+		return true
+	case parquet.FixedLenByteArray:
+		aBytes := a.ByteArray() // FixedLenByteArray also uses ByteArray() method
+		bBytes := b.ByteArray()
+		if len(aBytes) != len(bBytes) {
+			return false
+		}
+		for i, v := range aBytes {
+			if v != bBytes[i] {
+				return false
+			}
+		}
+		return true
+	default:
+		return false
+	}
+}
+
+// Benchmark tests
+func BenchmarkToParquetValue_BasicTypes(b *testing.B) {
+	value := &schema_pb.Value{
+		Kind: &schema_pb.Value_Int64Value{Int64Value: 12345678901234},
+	}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_, _ = toParquetValue(value)
+	}
+}
+
+func BenchmarkToParquetValue_TimestampValue(b *testing.B) {
+	value := &schema_pb.Value{
+		Kind: &schema_pb.Value_TimestampValue{
+			TimestampValue: &schema_pb.TimestampValue{
+				TimestampMicros: time.Now().UnixMicro(),
+				IsUtc:           true,
+			},
+		},
+	}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_, _ = toParquetValue(value)
+	}
+}
+
+func BenchmarkToParquetValue_DecimalValue(b *testing.B) {
+	value := &schema_pb.Value{
+		Kind: &schema_pb.Value_DecimalValue{
+			DecimalValue: &schema_pb.DecimalValue{
+				Value:     encodeBigIntToBytes(big.NewInt(123456789012345)),
+				Precision: 15,
+				Scale:     2,
+			},
+		},
+	}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_, _ = toParquetValue(value)
+	}
+}
--- a/weed/mq/schema/to_schema_value.go
+++ b/weed/mq/schema/to_schema_value.go
@@ -1,7 +1,9 @@
 package schema

 import (
+	"bytes"
 	"fmt"
+
 	"github.com/parquet-go/parquet-go"
 	"github.com/seaweedfs/seaweedfs/weed/pb/schema_pb"
 )
@@ -77,9 +79,68 @@ func toScalarValue(scalarType schema_pb.ScalarType, levels *ParquetLevels, value
 	case schema_pb.ScalarType_DOUBLE:
 		return &schema_pb.Value{Kind: &schema_pb.Value_DoubleValue{DoubleValue: value.Double()}}, valueIndex + 1, nil
 	case schema_pb.ScalarType_BYTES:
-		return &schema_pb.Value{Kind: &schema_pb.Value_BytesValue{BytesValue: value.ByteArray()}}, valueIndex + 1, nil
+		// Handle nil byte arrays from parquet to prevent growslice panic
+		byteData := value.ByteArray()
+		if byteData == nil {
+			byteData = []byte{} // Use empty slice instead of nil
+		}
+		return &schema_pb.Value{Kind: &schema_pb.Value_BytesValue{BytesValue: byteData}}, valueIndex + 1, nil
 	case schema_pb.ScalarType_STRING:
-		return &schema_pb.Value{Kind: &schema_pb.Value_StringValue{StringValue: string(value.ByteArray())}}, valueIndex + 1, nil
+		// Handle nil byte arrays from parquet to prevent string conversion issues
+		byteData := value.ByteArray()
+		if byteData == nil {
+			byteData = []byte{} // Use empty slice instead of nil
+		}
+		return &schema_pb.Value{Kind: &schema_pb.Value_StringValue{StringValue: string(byteData)}}, valueIndex + 1, nil
+	// Parquet logical types - convert from their physical storage back to logical values
+	case schema_pb.ScalarType_TIMESTAMP:
+		// Stored as INT64, convert back to TimestampValue
+		return &schema_pb.Value{
+			Kind: &schema_pb.Value_TimestampValue{
+				TimestampValue: &schema_pb.TimestampValue{
+					TimestampMicros: value.Int64(),
+					IsUtc:           true, // Default to UTC for compatibility
+				},
+			},
+		}, valueIndex + 1, nil
+	case schema_pb.ScalarType_DATE:
+		// Stored as INT32, convert back to DateValue
+		return &schema_pb.Value{
+			Kind: &schema_pb.Value_DateValue{
+				DateValue: &schema_pb.DateValue{
+					DaysSinceEpoch: value.Int32(),
+				},
+			},
+		}, valueIndex + 1, nil
+	case schema_pb.ScalarType_DECIMAL:
+		// Stored as FixedLenByteArray, convert back to DecimalValue
+		fixedBytes := value.ByteArray() // FixedLenByteArray also uses ByteArray() method
+		if fixedBytes == nil {
+			fixedBytes = []byte{} // Use empty slice instead of nil
+		}
+		// Remove leading zeros to get the minimal representation
+		trimmedBytes := bytes.TrimLeft(fixedBytes, "\x00")
+		if len(trimmedBytes) == 0 {
+			trimmedBytes = []byte{0} // Ensure we have at least one byte for zero
+		}
+		return &schema_pb.Value{
+			Kind: &schema_pb.Value_DecimalValue{
+				DecimalValue: &schema_pb.DecimalValue{
+					Value:     trimmedBytes,
+					Precision: 38, // Maximum precision supported by schema
+					Scale:     18, // Maximum scale supported by schema
+				},
+			},
+		}, valueIndex + 1, nil
+	case schema_pb.ScalarType_TIME:
+		// Stored as INT64, convert back to TimeValue
+		return &schema_pb.Value{
+			Kind: &schema_pb.Value_TimeValue{
+				TimeValue: &schema_pb.TimeValue{
+					TimeMicros: value.Int64(),
+				},
+			},
+		}, valueIndex + 1, nil
 	}
 	return nil, valueIndex, fmt.Errorf("unsupported scalar type: %v", scalarType)
 }
--- a/weed/mq/sub_coordinator/sub_coordinator.go
+++ b/weed/mq/sub_coordinator/sub_coordinator.go
@@ -2,6 +2,7 @@ package sub_coordinator

 import (
 	"fmt"
+
 	cmap "github.com/orcaman/concurrent-map/v2"
 	"github.com/seaweedfs/seaweedfs/weed/filer_client"
 	"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
--- a/weed/mq/topic/local_manager.go
+++ b/weed/mq/topic/local_manager.go
@@ -1,11 +1,12 @@
 package topic

 import (
+	"time"
+
 	cmap "github.com/orcaman/concurrent-map/v2"
 	"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/schema_pb"
 	"github.com/shirou/gopsutil/v3/cpu"
-	"time"
 )

 // LocalTopicManager manages topics on local broker
--- a/weed/mq/topic/local_partition.go
+++ b/weed/mq/topic/local_partition.go
@@ -3,6 +3,10 @@ package topic
 import (
 	"context"
 	"fmt"
+	"sync"
+	"sync/atomic"
+	"time"
+
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
@@ -10,9 +14,6 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
-	"sync"
-	"sync/atomic"
-	"time"
 )

 type LocalPartition struct {
--- a/weed/mq/topic/topic.go
+++ b/weed/mq/topic/topic.go
@@ -5,11 +5,14 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
+	"time"

 	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/schema_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 	jsonpb "google.golang.org/protobuf/encoding/protojson"
 )

@@ -102,3 +105,65 @@ func (t Topic) WriteConfFile(client filer_pb.SeaweedFilerClient, conf *mq_pb.Con
 	}
 	return nil
 }
+
+// DiscoverPartitions discovers all partition directories for a topic by scanning the filesystem
+// This centralizes partition discovery logic used across query engine, shell commands, etc.
+func (t Topic) DiscoverPartitions(ctx context.Context, filerClient filer_pb.FilerClient) ([]string, error) {
+	var partitionPaths []string
+
+	// Scan the topic directory for version directories (e.g., v2025-09-01-07-16-34)
+	err := filer_pb.ReadDirAllEntries(ctx, filerClient, util.FullPath(t.Dir()), "", func(versionEntry *filer_pb.Entry, isLast bool) error {
+		if !versionEntry.IsDirectory {
+			return nil // Skip non-directories
+		}
+
+		// Parse version timestamp from directory name (e.g., "v2025-09-01-07-16-34")
+		if !IsValidVersionDirectory(versionEntry.Name) {
+			// Skip directories that don't match the version format
+			return nil
+		}
+
+		// Scan partition directories within this version (e.g., 0000-0630)
+		versionDir := fmt.Sprintf("%s/%s", t.Dir(), versionEntry.Name)
+		return filer_pb.ReadDirAllEntries(ctx, filerClient, util.FullPath(versionDir), "", func(partitionEntry *filer_pb.Entry, isLast bool) error {
+			if !partitionEntry.IsDirectory {
+				return nil // Skip non-directories
+			}
+
+			// Parse partition boundary from directory name (e.g., "0000-0630")
+			if !IsValidPartitionDirectory(partitionEntry.Name) {
+				return nil // Skip invalid partition names
+			}
+
+			// Add this partition path to the list
+			partitionPath := fmt.Sprintf("%s/%s", versionDir, partitionEntry.Name)
+			partitionPaths = append(partitionPaths, partitionPath)
+			return nil
+		})
+	})
+
+	return partitionPaths, err
+}
+
+// IsValidVersionDirectory checks if a directory name matches the topic version format
+// Format: v2025-09-01-07-16-34
+func IsValidVersionDirectory(name string) bool {
+	if !strings.HasPrefix(name, "v") || len(name) != 20 {
+		return false
+	}
+
+	// Try to parse the timestamp part
+	timestampStr := name[1:] // Remove 'v' prefix
+	_, err := time.Parse("2006-01-02-15-04-05", timestampStr)
+	return err == nil
+}
+
+// IsValidPartitionDirectory checks if a directory name matches the partition boundary format
+// Format: 0000-0630 (rangeStart-rangeStop)
+func IsValidPartitionDirectory(name string) bool {
+	// Use existing ParsePartitionBoundary function to validate
+	start, stop := ParsePartitionBoundary(name)
+
+	// Valid partition ranges should have start < stop (and not both be 0, which indicates parse error)
+	return start < stop && start >= 0
+}