Flink : UT testPrimaryKeyFieldsAtEndOfTableSchema fails probablistcally

[singhpk234]

Test testPrimaryKeyFieldsAtEndOfTableSchema[catalogName=testhadoop, baseNamespace=default, format=PARQUET, isStreaming=true](org.apache.iceberg.flink.TestFlinkUpsert) failed with:
java.lang.AssertionError: 
Expecting:
  [+I[3, bbb, 2022-03-01], +I[1, aaa, 2022-03-01]]
to contain exactly in any order:
  [+I[2, aaa, 2022-03-01], +I[3, bbb, 2022-03-01]]
elements not found:
  [+I[2, aaa, 2022-03-01]]
and elements not expected:
  [+I[1, aaa, 2022-03-01]]

	at org.apache.iceberg.flink.TestHelpers.assertRows(TestHelpers.java:137)
	at org.apache.iceberg.flink.TestFlinkUpsert.testPrimaryKeyFieldsAtEndOfTableSchema(TestFlinkUpsert.java:262)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
	at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
	at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
	at org.apache.flink.util.TestNameProvider$1.evaluate(TestNameProvider.java:45)
	at org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:61)
	at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
	at org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100)
	at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366)
	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:103)
	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:63)
	at org.junit.runners.ParentRunner$4.run(ParentRunner.java:331)
	at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79)
	at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329)
	at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66)
	at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293)
	at org.junit.runners.ParentRunner.run(ParentRunner.java:413)
	at org.junit.runners.Suite.runChild(Suite.java:128)
	at org.junit.runners.Suite.runChild(Suite.java:27)
	at org.junit.runners.ParentRunner$4.run(ParentRunner.java:331)
	at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79)
	at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329)
	at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66)
	at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293)
	at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
	at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
	at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:54)
	at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:54)
	at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:54)
	at org.junit.rules.RunRules.evaluate(RunRules.java:20)
	at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
	at org.junit.runners.ParentRunner.run(ParentRunner.java:413)
	at org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.runTestClass(JUnitTestClassExecutor.java:110)
	at org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:58)
	at org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:38)
	at org.gradle.api.internal.tasks.testing.junit.AbstractJUnitTestClassProcessor.processTestClass(AbstractJUnitTestClassProcessor.java:62)
	at org.gradle.api.internal.tasks.testing.SuiteTestClassProcessor.processTestClass(SuiteTestClassProcessor.java:51)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:36)
	at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
	at org.gradle.internal.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:33)
	at org.gradle.internal.dispatch.ProxyDispatchAdapter$DispatchingInvocationHandler.invoke(ProxyDispatchAdapter.java:94)
	at com.sun.proxy.$Proxy2.processTestClass(Unknown Source)
	at org.gradle.api.internal.tasks.testing.worker.TestWorker$2.run(TestWorker.java:176)
	at org.gradle.api.internal.tasks.testing.worker.TestWorker.executeAndMaintainThreadName(TestWorker.java:129)
	at org.gradle.api.internal.tasks.testing.worker.TestWorker.execute(TestWorker.java:100)
	at org.gradle.api.internal.tasks.testing.worker.TestWorker.execute(TestWorker.java:60)
	at org.gradle.process.internal.worker.child.ActionExecutionWorker.execute(ActionExecutionWorker.java:56)
	at org.gradle.process.internal.worker.child.SystemApplicationClassLoaderWorker.call(SystemApplicationClassLoaderWorker.java:133)
	at org.gradle.process.internal.worker.child.SystemApplicationClassLoaderWorker.call(SystemApplicationClassLoaderWorker.java:71)
	at worker.org.gradle.process.internal.worker.GradleWorkerMain.run(GradleWorkerMain.java:69)
	at worker.org.gradle.process.internal.worker.GradleWorkerMain.main(GradleWorkerMain.java:74)

---

Looks like upsert is not working properly and hence causing flakyness of this UT

sample github run with failure : https://github.com/apache/iceberg/runs/5846777166?check_suite_focus=true

[openinx]

Yes, seems I can reproduce this failing parquet & streaming test cases sometimes in my host:

[openinx]

I opened a debug to verify those files, and I can see the files are laying like the following:

➜  default find . 
.
./default
./a.txt
./db
./db/upsert_on_pk_at_schema_end
./db/upsert_on_pk_at_schema_end/data
./db/upsert_on_pk_at_schema_end/data/data=aaa
./db/upsert_on_pk_at_schema_end/data/data=aaa/.00000-0-f9269f62-299b-4689-8b9f-ad59d7fb539d-00002.parquet.crc
./db/upsert_on_pk_at_schema_end/data/data=aaa/00000-0-f9269f62-299b-4689-8b9f-ad59d7fb539d-00002.parquet
./db/upsert_on_pk_at_schema_end/data/data=aaa/.00000-0-f9269f62-299b-4689-8b9f-ad59d7fb539d-00001.parquet.crc
./db/upsert_on_pk_at_schema_end/data/data=aaa/00000-0-f9269f62-299b-4689-8b9f-ad59d7fb539d-00001.parquet
./db/upsert_on_pk_at_schema_end/data/data=aaa/00000-0-f9269f62-299b-4689-8b9f-ad59d7fb539d-00005.parquet
./db/upsert_on_pk_at_schema_end/data/data=aaa/.00000-0-f9269f62-299b-4689-8b9f-ad59d7fb539d-00005.parquet.crc
./db/upsert_on_pk_at_schema_end/data/data=bbb
./db/upsert_on_pk_at_schema_end/data/data=bbb/00000-0-f9269f62-299b-4689-8b9f-ad59d7fb539d-00003.parquet
./db/upsert_on_pk_at_schema_end/data/data=bbb/.00000-0-f9269f62-299b-4689-8b9f-ad59d7fb539d-00004.parquet.crc
./db/upsert_on_pk_at_schema_end/data/data=bbb/00000-0-f9269f62-299b-4689-8b9f-ad59d7fb539d-00004.parquet
./db/upsert_on_pk_at_schema_end/data/data=bbb/.00000-0-f9269f62-299b-4689-8b9f-ad59d7fb539d-00003.parquet.crc
./db/upsert_on_pk_at_schema_end/metadata
./db/upsert_on_pk_at_schema_end/metadata/version-hint.text
./db/upsert_on_pk_at_schema_end/metadata/.version-hint.text.crc
./db/upsert_on_pk_at_schema_end/metadata/e21484e0-cea6-4001-b4af-3e34c9249a88-m0.avro
./db/upsert_on_pk_at_schema_end/metadata/snap-3867050424845709517-1-e21484e0-cea6-4001-b4af-3e34c9249a88.avro
./db/upsert_on_pk_at_schema_end/metadata/e21484e0-cea6-4001-b4af-3e34c9249a88-m1.avro
./db/upsert_on_pk_at_schema_end/metadata/.v2.metadata.json.crc
./db/upsert_on_pk_at_schema_end/metadata/v2.metadata.json
./db/upsert_on_pk_at_schema_end/metadata/.v1.metadata.json.crc
./db/upsert_on_pk_at_schema_end/metadata/.snap-3867050424845709517-1-e21484e0-cea6-4001-b4af-3e34c9249a88.avro.crc
./db/upsert_on_pk_at_schema_end/metadata/.e21484e0-cea6-4001-b4af-3e34c9249a88-m1.avro.crc
./db/upsert_on_pk_at_schema_end/metadata/.e21484e0-cea6-4001-b4af-3e34c9249a88-m0.avro.crc
./db/upsert_on_pk_at_schema_end/metadata/v1.metadata.json

That means there is only one checkpoint to commit those three records, while in the partition data=aaa, those records are:

#  The equality delete file.
➜  default parquet cat ./db/upsert_on_pk_at_schema_end/data/data=aaa/00000-0-f9269f62-299b-4689-8b9f-ad59d7fb539d-00002.parquet 
{"data": "aaa", "dt": 19052}

# The insert data file.
➜  default parquet cat ./db/upsert_on_pk_at_schema_end/data/data=aaa/00000-0-f9269f62-299b-4689-8b9f-ad59d7fb539d-00001.parquet 
{"id": 2, "data": "aaa", "dt": 19052}
{"id": 1, "data": "aaa", "dt": 19052}

# The positional delete file.
➜  default parquet cat ./db/upsert_on_pk_at_schema_end/data/data=aaa/00000-0-f9269f62-299b-4689-8b9f-ad59d7fb539d-00005.parquet
{"file_path": "file:/var/folders/fg/kbb4swcd0gl_3s0wlhhk9bch0000gp/T/junit3828094481211623109/default/db/upsert_on_pk_at_schema_end/data/data=aaa/00000-0-f9269f62-299b-4689-8b9f-ad59d7fb539d-00001.parquet", "pos": 0}

The tricky thing is: the record (2, 'aaa', '2022-03-1') was written before record(1, 'aaa', '2022-03-01'), that' why we encountered the failure case.

[yittg]

Another case failed in testPrimaryKeyFieldsAtBeginningOfSchema

org.apache.iceberg.flink.TestFlinkUpsert > testPrimaryKeyFieldsAtBeginningOfSchema[catalogName=testhadoop, baseNamespace=default, format=PARQUET, isStreaming=true] FAILED
    java.lang.AssertionError: 
    Expecting:
      [+I[bbb, 2022-03-01, 3], +I[aaa, 2022-03-01, 1]]
    to contain exactly in any order:
      [+I[aaa, 2022-03-01, 2], +I[bbb, 2022-03-01, 3]]
    elements not found:
      [+I[aaa, 2022-03-01, 2]]
    and elements not expected:
      [+I[aaa, 2022-03-01, 1]]
        at org.apache.iceberg.flink.TestHelpers.assertRows(TestHelpers.java:137)
        at org.apache.iceberg.flink.TestFlinkUpsert.testPrimaryKeyFieldsAtBeginningOfSchema(TestFlinkUpsert.java:[219](https://github.com/apache/iceberg/runs/5834902185?check_suite_focus=true#step:6:219))

[yittg]

After some investigate, i found the Values is not really one operator node with tuple list, instead it derived three parallel Calc nodes due to some cast.
So the order can not be guaranteed.

[yittg]

So let's split it to two statements if we still use the same way for creating data.

WDYT, @openinx

[openinx]

After some investigate, i found the Values is not really one operator node with tuple list, instead it derived three parallel Calc nodes due to some cast.

Could you please provide the flink code reference where we dispatched the values into multiple parallelism ? Actually, I don't find the place...

[yittg]

@openinx , actually it happens in calcite rel converter.
Only pure literal tuples can be converted to LogicalValues node, otherwise it will be converted to multiple LogicalProject -> LogicalUnion nodes.

Change TO_DATE('xxxx-xx-xx') to DATE 'xxxx-xx-xx' can make it a literal node and avoid being converted to union nodes.

FYR:

https://github.com/apache/calcite/blob/32babcb90aa5940bdd75760021abab6459a9cca4/core/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java#L4472-L4475

https://github.com/apache/calcite/blob/32babcb90aa5940bdd75760021abab6459a9cca4/core/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java#L1733-L1741

[kbendick]

@yittg this is great work finding this and fixing it. I am just getting back from being out of office, so am catching up on this now.

Thank you to @singhpk234 and @openinx as well for all of the work on this. Really appreciate your thorough analysis.