You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "Kaixuan-Duan (via GitHub)" <gi...@apache.org> on 2026/04/28 13:54:16 UTC
[PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Kaixuan-Duan opened a new pull request, #3227:
URL: https://github.com/apache/fluss/pull/3227
### Purpose
Linked issue: close #2724
When Spark pushes down an empty column projection for `COUNT(*)`/`COUNT(1)` queries, the Fluss server fails with `IllegalStateException("Invalid metadata length")` in `FileLogProjection.project()`, causing the client to retry indefinitely and the query to hang.
This PR fixes the issue from two sides:
Server side: reject empty projection early with a clear `InvalidColumnProjectionException` instead of crashing with an internal error.
Spark connector side: fall back to projecting the first column when Spark pushes down an empty projection, so the row count is preserved without fetching unnecessary data.
### Brief change log
- `FileLogProjection#setCurrentProjection`: add a guard that throws InvalidColumnProjectionException when selectedFieldPositions is empty.
- `FileLogProjectionTest`: add testEmptyProjectionRejectsWithClearError to verify the server-side guard.
- `FlussBatch#projection` / `FlussMicroBatchStream#projection`: when readSchema yields an empty projection, fall back to Array(0) (first column).
- `SparkLogTableReadTest`: add `COUNT(*)` and `COUNT(1)` end-to-end tests for log tables.
- `SparkPrimaryKeyTableReadTest`: add `COUNT(*)` end-to-end test for primary key tables.
### Tests
./mvnw -pl fluss-common -am -DskipTests=false -Dtest=FileLogProjectionTest#testEmptyProjectionRejectsWithClearError test
./mvnw -pl fluss-spark/fluss-spark-ut -am install -DskipTests
./mvnw -pl fluss-spark/fluss-spark-ut -Dsuites='org.apache.fluss.spark.SparkLogTableReadTest' test
./mvnw -pl fluss-spark/fluss-spark-ut -Dsuites='org.apache.fluss.spark.SparkPrimaryKeyTableReadTest' test
### API and Format
### Documentation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Posted by "Kaixuan-Duan (via GitHub)" <gi...@apache.org>.
Kaixuan-Duan commented on PR #3227:
URL: https://github.com/apache/fluss/pull/3227#issuecomment-4354236676
@luoyuxia I’m sorry. It was my fault. I didn't notice that there was already a PR pointing to issue#2724
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Posted by "beryllw (via GitHub)" <gi...@apache.org>.
beryllw commented on PR #3227:
URL: https://github.com/apache/fluss/pull/3227#issuecomment-4484938614
> @beryllw curious could we assume table statistics always accurate in fluss?
I think we could assume they're always accurate in Fluss — for KV tables it's backed by rowCount, and for log tables it's derived from getHighWatermark() - logStartOffset(), which should give us a reliable count.
https://github.com/apache/fluss/blob/d4cd1a2771ff254a81877296a5cabfa966d102e2/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java#L248-L250
https://github.com/apache/fluss/blob/c6a1c2dbfe36aa8b503cff04470acf2951ef9e51/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java#L307-L318
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #3227:
URL: https://github.com/apache/fluss/pull/3227#issuecomment-4349394067
@Kaixuan-Duan Hi, is it same with #2725. cc @beryllw
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Posted by "Kaixuan-Duan (via GitHub)" <gi...@apache.org>.
Kaixuan-Duan commented on PR #3227:
URL: https://github.com/apache/fluss/pull/3227#issuecomment-4951242632
@luoyuxia Thanks for the suggestion. I've split this PR into a pure bug fix — the aggregation pushdown part has been removed and will be submitted as a follow-up PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Posted by "Kaixuan-Duan (via GitHub)" <gi...@apache.org>.
Kaixuan-Duan commented on code in PR #3227:
URL: https://github.com/apache/fluss/pull/3227#discussion_r3635085046
##########
fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkPrimaryKeyTableReadTest.scala:
##########
@@ -135,6 +135,32 @@ class SparkPrimaryKeyTableReadTest extends FlussSparkTestBase {
Row(800L, 23L, "addr3") ::
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(*) — empty projection case (regression test for #2724). PK table has 7 rows
Review Comment:
You're right. Verified locally by disabling the fallback and the test still passed. Removed the redundant PK table tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #3227:
URL: https://github.com/apache/fluss/pull/3227#discussion_r3619217513
##########
fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkPrimaryKeyTableReadTest.scala:
##########
@@ -135,6 +135,32 @@ class SparkPrimaryKeyTableReadTest extends FlussSparkTestBase {
Row(800L, 23L, "addr3") ::
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(*) — empty projection case (regression test for #2724). PK table has 7 rows
Review Comment:
nit: FlussUpsertPartitionReader always appends pk columns (projectionWithPks), so the PK path never actually sent an empty projection to the server, so does this test pass without the fix too?
##########
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussBatch.scala:
##########
@@ -49,12 +49,15 @@ abstract class FlussBatch(
protected def projection: Array[Int] = {
val columnNameToIndex = tableInfo.getSchema.getColumnNames.asScala.zipWithIndex.toMap
- readSchema.fields.map {
+ val projected = readSchema.fields.map {
field =>
columnNameToIndex.getOrElse(
field.name,
throw new IllegalArgumentException(s"Invalid field name: ${field.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
Review Comment:
projectionOf is the third copy of this name->index resolution (came in with #3559, which pointed only the planners at it), and it still returns empty, which the planners feed to lakeSource.withProject().
Shall we put the fallback in projectionOf once and have FlussBatch/FlussMicroBatchStream.projection delegate to it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Posted by "beryllw (via GitHub)" <gi...@apache.org>.
beryllw commented on code in PR #3227:
URL: https://github.com/apache/fluss/pull/3227#discussion_r3207161287
##########
fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkPrimaryKeyTableReadTest.scala:
##########
@@ -376,4 +376,22 @@ class SparkPrimaryKeyTableReadTest extends FlussSparkTestBase {
assert(numRowsRead == 3L, s"Expected 3 rows read, got $numRowsRead")
}
}
+
+ test("Spark Read: primary key table COUNT(*) without filter") {
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni merged PR #3227:
URL: https://github.com/apache/fluss/pull/3227
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Posted by "Yohahaha (via GitHub)" <gi...@apache.org>.
Yohahaha commented on PR #3227:
URL: https://github.com/apache/fluss/pull/3227#issuecomment-4417009840
> I've closed PR #2725.Please go ahead. Also, is it possible to use the pre-computed table statistics directly when dealing with PK tables, instead of scanning the data and counting afterward?
@beryllw curious could we assume table statistics always accurate in fluss?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Posted by "Yohahaha (via GitHub)" <gi...@apache.org>.
Yohahaha commented on code in PR #3227:
URL: https://github.com/apache/fluss/pull/3227#discussion_r3165395914
##########
fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkLogTableReadTest.scala:
##########
@@ -451,6 +451,20 @@ class SparkLogTableReadTest extends FlussSparkTestBase {
}
}
+ test("Spark Read: COUNT(*) without filter") {
Review Comment:
What happens when there is a filter?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Posted by "Kaixuan-Duan (via GitHub)" <gi...@apache.org>.
Kaixuan-Duan commented on code in PR #3227:
URL: https://github.com/apache/fluss/pull/3227#discussion_r3169483404
##########
fluss-common/src/main/java/org/apache/fluss/record/FileLogProjection.java:
##########
@@ -121,6 +121,19 @@ public void setCurrentProjection(
SchemaGetter schemaGetter,
ArrowCompressionInfo compressionInfo,
int[] selectedFieldPositions) {
+ // Empty projection (selectedFieldPositions.length == 0) is currently not supported on the
Review Comment:
Looked into the Flink connector: it doesn't implement SupportsAggregatePushDown, and Flink's planner doesn't push an empty projection for COUNT(*) either, so this bug is not reproducible there today. The new server-side guard is connector-agnostic, so it still protects Flink/Kafka/direct API users if any caller ever pushes an empty projection in the future.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Posted by "Kaixuan-Duan (via GitHub)" <gi...@apache.org>.
Kaixuan-Duan commented on code in PR #3227:
URL: https://github.com/apache/fluss/pull/3227#discussion_r3169494435
##########
fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkLogTableReadTest.scala:
##########
@@ -451,6 +451,20 @@ class SparkLogTableReadTest extends FlussSparkTestBase {
}
}
+ test("Spark Read: COUNT(*) without filter") {
Review Comment:
COUNT(*) with filter will not traverse an empty projection path. The original behavior is correct, and this PR does not affect it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #3227:
URL: https://github.com/apache/fluss/pull/3227#issuecomment-4881744443
@Kaixuan-Duan Thanks for fixing the empty projection issue. My main concern is the newly added Spark `COUNT` pushdown: before merging, I think it should be aligned with the existing Flink count-pushdown behavior.
Flink already has a clear contract in `FlinkTableSource#applyAggregates` / `PushdownUtils.countTable`: only push down a single `COUNT(*)` / `COUNT(1)` / `COUNT(non-null column)` without grouping; do not push down nullable `COUNT(col)`, streaming reads, or lake-enabled reads; and let WAL / legacy PK tables with disabled row count fail with the existing row-count-disabled error.
The Spark implementation is partially aligned, but a few cases still need to be clarified or covered:
- WAL / legacy row-count-disabled PK tables are not covered by Spark tests. If failure is intended to match Flink, please add an explicit test.
- Spark should explicitly avoid aggregate pushdown for streaming reads, matching Flink's `streaming` guard, or add coverage proving Spark will never call this path for streaming.
- `FlussCountScan` bypasses the existing Spark PK batch scan startup-mode restriction: normal PK batch reads only support `FULL`, but this fast path can return a result under other startup modes.
I think the empty-projection fix can be kept, but Spark `COUNT` pushdown should either be aligned with Flink's contract and tested for these edge cases, or be split out from this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Posted by "Kaixuan-Duan (via GitHub)" <gi...@apache.org>.
Kaixuan-Duan commented on PR #3227:
URL: https://github.com/apache/fluss/pull/3227#issuecomment-5011284645
@luoyuxia Thanks，I have rebased the branch onto the main.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Posted by "beryllw (via GitHub)" <gi...@apache.org>.
beryllw commented on code in PR #3227:
URL: https://github.com/apache/fluss/pull/3227#discussion_r3207158511
##########
fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkLogTableReadTest.scala:
##########
@@ -451,6 +451,20 @@ class SparkLogTableReadTest extends FlussSparkTestBase {
}
}
+ test("Spark Read: COUNT(*) without filter") {
Review Comment:
Perhaps we could add the COUNT(*) test case to the existing test("Spark Read: log table") rather than introducing a new test.
Additionally, we should test COUNT(*) with filters and nullable column types. It would also be worth testing against partitioned tables.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Posted by "Kaixuan-Duan (via GitHub)" <gi...@apache.org>.
Kaixuan-Duan commented on code in PR #3227:
URL: https://github.com/apache/fluss/pull/3227#discussion_r3635083745
##########
fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussBatch.scala:
##########
@@ -49,12 +49,15 @@ abstract class FlussBatch(
protected def projection: Array[Int] = {
val columnNameToIndex = tableInfo.getSchema.getColumnNames.asScala.zipWithIndex.toMap
- readSchema.fields.map {
+ val projected = readSchema.fields.map {
field =>
columnNameToIndex.getOrElse(
field.name,
throw new IllegalArgumentException(s"Invalid field name: ${field.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
Review Comment:
Good catch — done. Moved the fallback into FlussScanBuilder.projectionOf and made FlussBatch/FlussMicroBatchStream.projection delegate to it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Posted by "beryllw (via GitHub)" <gi...@apache.org>.
beryllw commented on PR #3227:
URL: https://github.com/apache/fluss/pull/3227#issuecomment-4404580864
I've closed PR #2725.Please go ahead.
Also, is it possible to use the pre-computed table statistics directly when dealing with PK tables, instead of scanning the data and counting afterward?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Posted by "Yohahaha (via GitHub)" <gi...@apache.org>.
Yohahaha commented on code in PR #3227:
URL: https://github.com/apache/fluss/pull/3227#discussion_r3165366906
##########
fluss-common/src/main/java/org/apache/fluss/record/FileLogProjection.java:
##########
@@ -121,6 +121,19 @@ public void setCurrentProjection(
SchemaGetter schemaGetter,
ArrowCompressionInfo compressionInfo,
int[] selectedFieldPositions) {
+ // Empty projection (selectedFieldPositions.length == 0) is currently not supported on the
Review Comment:
It would be good to also verify this behavior and the fix in the Flink connector.
##########
fluss-common/src/main/java/org/apache/fluss/record/FileLogProjection.java:
##########
@@ -121,6 +121,19 @@ public void setCurrentProjection(
SchemaGetter schemaGetter,
ArrowCompressionInfo compressionInfo,
int[] selectedFieldPositions) {
+ // Empty projection (selectedFieldPositions.length == 0) is currently not supported on the
Review Comment:
It would be good to also verify this behavior and the fix in the Flink connector if needed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Posted by "Kaixuan-Duan (via GitHub)" <gi...@apache.org>.
Kaixuan-Duan commented on PR #3227:
URL: https://github.com/apache/fluss/pull/3227#issuecomment-4412818377
@beryllw Thanks for the suggestion! Done. The Spark connector now implements `SupportsPushDownAggregates` and answers `COUNT(*)` / `COUNT(1)` / `COUNT(non-null col)` (no GROUP BY, no DISTINCT) on PK non-lake tables directly via `Admin.getTableStats().getRowCount()` — no data scan. Falls back gracefully on `UnsupportedVersionException` for older clusters. PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Fix empty projection causing Invalid metadata length for COUNT(*)/COUNT(1) [fluss]
Posted by "Kaixuan-Duan (via GitHub)" <gi...@apache.org>.
Kaixuan-Duan commented on PR #3227:
URL: https://github.com/apache/fluss/pull/3227#issuecomment-5053503563
@fresh-borzoni Thanks for the suggestion. Projection.of now rejects empty arrays eagerly. PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org