You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ab...@apache.org on 2026/07/23 10:14:13 UTC
(fluss) branch main updated: [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) (#3227)
This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new cf3b7b8de [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) (#3227)
cf3b7b8de is described below
commit cf3b7b8de176f2ede381092a13984cce8957b36b
Author: Kaixuan Duan <98...@users.noreply.github.com>
AuthorDate: Thu Jul 23 18:14:09 2026 +0800
[spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) (#3227)
* [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1)
* Address feedback
---
.../org/apache/fluss/record/FileLogProjection.java | 12 ++++++++
.../java/org/apache/fluss/utils/Projection.java | 4 +++
.../apache/fluss/record/FileLogProjectionTest.java | 34 ++++++++++++++++++++++
.../org/apache/fluss/utils/ProjectionTest.java | 11 +++++++
.../org/apache/fluss/spark/read/FlussBatch.scala | 13 ++-------
.../fluss/spark/read/FlussMicroBatchStream.scala | 11 ++-----
.../apache/fluss/spark/read/FlussScanBuilder.scala | 5 +++-
.../apache/fluss/spark/SparkLogTableReadTest.scala | 33 +++++++++++++++++++++
.../spark/lake/SparkLakeLogTableReadTest.scala | 17 +++++++++++
9 files changed, 119 insertions(+), 21 deletions(-)
diff --git a/fluss-common/src/main/java/org/apache/fluss/record/FileLogProjection.java b/fluss-common/src/main/java/org/apache/fluss/record/FileLogProjection.java
index bedaac475..35f4202e0 100644
--- a/fluss-common/src/main/java/org/apache/fluss/record/FileLogProjection.java
+++ b/fluss-common/src/main/java/org/apache/fluss/record/FileLogProjection.java
@@ -121,6 +121,18 @@ public class FileLogProjection {
SchemaGetter schemaGetter,
ArrowCompressionInfo compressionInfo,
int[] selectedFieldPositions) {
+ // Empty projection is currently not supported on the server side: the Arrow metadata length
+ // estimate disagrees with the serialized length for a zero-field schema, which would
+ // cause the client to retry indefinitely. Fail fast with a clear, non-retriable error
+ // instead.
+ if (selectedFieldPositions != null && selectedFieldPositions.length == 0) {
+ throw new InvalidColumnProjectionException(
+ "Empty projection is not supported. "
+ + "Please project at least one column. "
+ + "For aggregations like COUNT(*) / COUNT(1), "
+ + "the connector should push down a projection that selects "
+ + "at least one column.");
+ }
this.tableId = tableId;
this.schemaGetter = schemaGetter;
this.compressionInfo = compressionInfo;
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/Projection.java b/fluss-common/src/main/java/org/apache/fluss/utils/Projection.java
index c16c6b10b..5f5a84f13 100644
--- a/fluss-common/src/main/java/org/apache/fluss/utils/Projection.java
+++ b/fluss-common/src/main/java/org/apache/fluss/utils/Projection.java
@@ -21,6 +21,7 @@ import org.apache.fluss.types.RowType;
import java.util.Arrays;
+import static org.apache.fluss.utils.Preconditions.checkArgument;
import static org.apache.fluss.utils.Preconditions.checkState;
/**
@@ -51,6 +52,9 @@ public class Projection {
/** Create a {@link Projection} of the provided {@code indexes}. */
public static Projection of(int[] indexes) {
+ checkArgument(
+ indexes != null && indexes.length > 0,
+ "Projection must select at least one column, but got an empty projection.");
return new Projection(indexes);
}
diff --git a/fluss-common/src/test/java/org/apache/fluss/record/FileLogProjectionTest.java b/fluss-common/src/test/java/org/apache/fluss/record/FileLogProjectionTest.java
index bacd2d1ab..1fb197c01 100644
--- a/fluss-common/src/test/java/org/apache/fluss/record/FileLogProjectionTest.java
+++ b/fluss-common/src/test/java/org/apache/fluss/record/FileLogProjectionTest.java
@@ -164,6 +164,40 @@ class FileLogProjectionTest {
"The projection indexes should not contain duplicated fields, but is [0, 0, 0]");
}
+ /**
+ * When the client pushes down an empty column projection (selectedFieldPositions = []), e.g.
+ * for {@code COUNT(*)} / {@code COUNT(1)} aggregations, the server should reject it eagerly
+ * with a clear {@link InvalidColumnProjectionException} instead of failing with an internal
+ * {@code IllegalStateException("Invalid metadata length")} that callers would retry
+ * indefinitely.
+ */
+ @Test
+ void testEmptyProjectionRejectsWithClearError() throws Exception {
+ long tableId = 1L;
+ short schemaId = (short) 2;
+ FileLogRecords recordsOfData2RowType =
+ createFileLogRecords(
+ schemaId,
+ LOG_MAGIC_VALUE_V1,
+ TestData.DATA2_ROW_TYPE,
+ TestData.DATA2,
+ TestData.DATA2);
+ FileLogProjection projection = new FileLogProjection(new ProjectionPushdownCache());
+
+ // Empty projection - emulates Spark COUNT(*)/COUNT(1) optimisation.
+ assertThatThrownBy(
+ () ->
+ doProjection(
+ tableId,
+ schemaId,
+ projection,
+ recordsOfData2RowType,
+ new int[] {},
+ recordsOfData2RowType.sizeInBytes()))
+ .isInstanceOf(InvalidColumnProjectionException.class)
+ .hasMessageContaining("Empty projection is not supported");
+ }
+
@Test
void testProjectionOldDataWithNewSchema() throws Exception {
// Currently, we only support add column at last.
diff --git a/fluss-common/src/test/java/org/apache/fluss/utils/ProjectionTest.java b/fluss-common/src/test/java/org/apache/fluss/utils/ProjectionTest.java
index 4f85e9516..efbf1768c 100644
--- a/fluss-common/src/test/java/org/apache/fluss/utils/ProjectionTest.java
+++ b/fluss-common/src/test/java/org/apache/fluss/utils/ProjectionTest.java
@@ -26,10 +26,21 @@ import org.apache.fluss.types.RowType;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link org.apache.fluss.utils.Projection}. */
class ProjectionTest {
+ @Test
+ void testEmptyProjectionIsRejected() {
+ assertThatThrownBy(() -> Projection.of(new int[0]))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Projection must select at least one column");
+ assertThatThrownBy(() -> Projection.of(null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Projection must select at least one column");
+ }
+
@Test
void testProjection() {
Projection projection = Projection.of(new int[] {2, 0, 3});
diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussBatch.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussBatch.scala
index ecc29326f..feade4ced 100644
--- a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussBatch.scala
+++ b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussBatch.scala
@@ -26,8 +26,6 @@ import org.apache.spark.sql.connector.read.{Batch, InputPartition, PartitionRead
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
-import scala.collection.JavaConverters._
-
/**
* Base class for planner-backed batch scans. The planner (constructed at ScanBuilder-time) opens
* and releases the Fluss client Connection/Admin itself — scoped to its `plan()` call — and
@@ -47,15 +45,8 @@ abstract class FlussBatch(
planner: SplitPlanner)
extends Batch {
- protected def projection: Array[Int] = {
- val columnNameToIndex = tableInfo.getSchema.getColumnNames.asScala.zipWithIndex.toMap
- readSchema.fields.map {
- field =>
- columnNameToIndex.getOrElse(
- field.name,
- throw new IllegalArgumentException(s"Invalid field name: ${field.name}"))
- }
- }
+ protected def projection: Array[Int] =
+ FlussScanBuilder.projectionOf(tableInfo, Some(readSchema))
override def planInputPartitions(): Array[InputPartition] = planner.plan()
}
diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussMicroBatchStream.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussMicroBatchStream.scala
index f7bf6722e..6351dba48 100644
--- a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussMicroBatchStream.scala
+++ b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussMicroBatchStream.scala
@@ -70,15 +70,8 @@ abstract class FlussMicroBatchStream(
val stoppingOffsetsInitializer: OffsetsInitializer =
FlussOffsetInitializers.stoppingOffsetsInitializer(false, options, flussConfig)
- protected def projection: Array[Int] = {
- val columnNameToIndex = tableInfo.getSchema.getColumnNames.asScala.zipWithIndex.toMap
- readSchema.fields.map {
- field =>
- columnNameToIndex.getOrElse(
- field.name,
- throw new IllegalArgumentException(s"Invalid field name: ${field.name}"))
- }
- }
+ protected def projection: Array[Int] =
+ FlussScanBuilder.projectionOf(tableInfo, Some(readSchema))
override def close(): Unit = {
if (admin != null) {
diff --git a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussScanBuilder.scala b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussScanBuilder.scala
index fe7d60e86..3ce15d1b0 100644
--- a/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussScanBuilder.scala
+++ b/fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussScanBuilder.scala
@@ -200,7 +200,7 @@ object FlussScanBuilder {
/** Convert a Spark required-schema projection back to Fluss column indices. */
def projectionOf(tableInfo: TableInfo, requiredSchema: Option[StructType]): Array[Int] = {
val allFields = (0 until tableInfo.getRowType.getFieldCount).toArray
- requiredSchema match {
+ val projected = requiredSchema match {
case None => allFields
case Some(schema) =>
val columnNameToIndex =
@@ -212,5 +212,8 @@ object FlussScanBuilder {
throw new IllegalArgumentException(s"Invalid field name: ${f.name}"))
}
}
+ // The Fluss server does not currently support empty Arrow projections. Fall back to projecting
+ // the first column so the row count is preserved while still avoiding fetching unnecessary columns.
+ if (projected.isEmpty) Array(0) else projected
}
}
diff --git a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkLogTableReadTest.scala b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkLogTableReadTest.scala
index 753e265ba..123f2914c 100644
--- a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkLogTableReadTest.scala
+++ b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkLogTableReadTest.scala
@@ -84,6 +84,30 @@ class SparkLogTableReadTest extends FlussSparkTestBase {
Row(1000L, 25L) ::
Row(1100L, 260L) :: Nil
)
+
+ // Insert one row with a NULL nullable column to make the COUNT(nullable_col)
+ // assertion below actually discriminate between row count and non-null count.
+ sql(s"""
+ |INSERT INTO $DEFAULT_DATABASE.t VALUES
+ |(1200L, 270L, 607, NULL)
+ |""".stripMargin)
+
+ // COUNT(*) / COUNT(1) — empty projection case
+ checkAnswer(sql(s"SELECT COUNT(*) FROM $DEFAULT_DATABASE.t"), Row(9L) :: Nil)
+ checkAnswer(sql(s"SELECT COUNT(1) FROM $DEFAULT_DATABASE.t"), Row(9L) :: Nil)
+
+ // COUNT(*) with filter — verifies the non-empty-projection path is unaffected
+ checkAnswer(
+ sql(s"SELECT COUNT(*) FROM $DEFAULT_DATABASE.t WHERE orderId >= 900"),
+ Row(5L) :: Nil
+ )
+
+ // COUNT on a nullable column — address is nullable STRING and one row has NULL,
+ // so COUNT(address) must be strictly less than COUNT(*).
+ checkAnswer(
+ sql(s"SELECT COUNT(address) FROM $DEFAULT_DATABASE.t"),
+ Row(8L) :: Nil
+ )
}
}
@@ -130,6 +154,15 @@ class SparkLogTableReadTest extends FlussSparkTestBase {
Row(800L, "addr3", "2026-01-02") ::
Row(900L, "addr4", "2026-01-02") :: Nil
)
+
+ // COUNT(*) on partitioned log table
+ checkAnswer(sql(s"SELECT COUNT(*) FROM $DEFAULT_DATABASE.t"), Row(5L) :: Nil)
+
+ // COUNT(*) with partition filter
+ checkAnswer(
+ sql(s"SELECT COUNT(*) FROM $DEFAULT_DATABASE.t WHERE dt = '2026-01-01'"),
+ Row(2L) :: Nil
+ )
}
}
diff --git a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakeLogTableReadTest.scala b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakeLogTableReadTest.scala
index 5fd910968..7d525db30 100644
--- a/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakeLogTableReadTest.scala
+++ b/fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakeLogTableReadTest.scala
@@ -117,6 +117,10 @@ abstract class SparkLakeLogTableReadTest extends SparkLakeTableReadTestBase {
sql(s"SELECT name FROM $DEFAULT_DATABASE.t_lake_only ORDER BY name"),
Row("alpha") :: Row("beta") :: Row("gamma") :: Nil
)
+
+ // COUNT(*) / COUNT(1) over a lake-only read: empty projection must fall back.
+ checkAnswer(sql(s"SELECT COUNT(*) FROM $DEFAULT_DATABASE.t_lake_only"), Row(3L) :: Nil)
+ checkAnswer(sql(s"SELECT COUNT(1) FROM $DEFAULT_DATABASE.t_lake_only"), Row(3L) :: Nil)
}
// Test partitioned table
@@ -152,6 +156,19 @@ abstract class SparkLakeLogTableReadTest extends SparkLakeTableReadTestBase {
Row("beta", "2026-01-01") ::
Row("gamma", "2026-01-02") :: Nil
)
+
+ // COUNT(*) over a partitioned lake-only read.
+ checkAnswer(
+ sql(s"SELECT COUNT(*) FROM $DEFAULT_DATABASE.t_lake_only_partitioned"),
+ Row(3L) :: Nil
+ )
+
+ // COUNT(*) with a partition filter.
+ checkAnswer(
+ sql(
+ s"SELECT COUNT(*) FROM $DEFAULT_DATABASE.t_lake_only_partitioned WHERE dt = '2026-01-01'"),
+ Row(2L) :: Nil
+ )
}
}