[Kernel]Add simple crc post commit for incremental crc writing. #4134

huan233usc · 2025-02-08T00:01:28Z

Which Delta project/connector is this regarding?

Description

This PR introduces a new post commit hook - ChecksumSimple, for writing CRC file after txn commit.
CRC file will only be written only commit version - 1's snapshot reads CRC during state construction

Other case will be handled in a separate PR

How was this patch tested?

E2e test

Does this PR introduce any user-facing changes?

No

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableWritesSuite.scala

scottsand-db

Looks good! Left some feedback on some of the code comments + organziation. Thanks!

Also -- please update the PR description.

kernel/kernel-api/src/main/java/io/delta/kernel/hook/PostCommitHook.java

kernel/kernel-api/src/main/java/io/delta/kernel/internal/TransactionImpl.java

kernel/kernel-api/src/main/java/io/delta/kernel/internal/checksum/ChecksumUtils.java

kernel/kernel-api/src/main/java/io/delta/kernel/internal/checksum/ChecksumWriter.java

scottsand-db · 2025-02-12T18:04:07Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/checksum/ChecksumWriter.java

+
+  private final Path logPath;
+  // Constants for schema field names
+  private static final String TABLE_SIZE_BYTES = "tableSizeBytes";


static fields should come before member fields

kernel/kernel-api/src/main/java/io/delta/kernel/internal/checksum/ChecksumWriter.java

kernel/kernel-api/src/test/scala/io/delta/kernel/internal/checksum/ChecksumWriterSuite.scala

.../kernel-defaults/src/test/scala/io/delta/kernel/defaults/ChecksumSimpleComparisonSuite.scala

scottsand-db · 2025-02-12T18:21:43Z

.../kernel-defaults/src/test/scala/io/delta/kernel/defaults/ChecksumSimpleComparisonSuite.scala

+    }
+  }
+
+  test("create table as select and verify checksum") {


sorry - what's the benefit of testing "create table as select" vs just "create table and insert to"?

are you expecting the CRC to be/look different between the two cases?

The only difference is the 00000.crc, ctas has added file while create table doesn't, so I added this test for completeness.

Removed the CTAS case, we believe it is a bit redundant here.

scottsand-db · 2025-02-12T18:22:09Z

.../kernel-defaults/src/test/scala/io/delta/kernel/defaults/ChecksumSimpleComparisonSuite.scala

+        .withSchema(engine, new StructType().add("id", INTEGER))
+        .build(engine)
+
+      copyAndCommitTxnForUnpartitionedTable(txn, engine, sparkTablePath, versionAtCommit = 0)


I couldn't quite understand what this method is doing -- can you explain to me (on this PR, not in the code)?

I was trying to make the test logic like

1.commit to spark table using ctas/dml, which produces a log with add file
2.read the log in step 1 and commit to the kernel table (make sore both table's log pointing to the same sets of parquet file to validate table size is correct in crc)
3.verify crc written stats are aligned

This method is mainly for step 2.

Does "copyDeltaLogFromSparkToKernel" a better name?

Added some more documentation to the test class and method as follow up

allisonport-db · 2025-02-19T21:01:33Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/TransactionImpl.java

+
+    return readSnapshot
+        .getCurrentCrcInfo()
+        .filter(lastCrcInfo -> commitAtVersion == lastCrcInfo.getVersion() + 1)


nit: want to add a comment like // in the case of a conflicting txn and successful retry the readSnapshot may not be commitVersion - 1

When I first saw this it was intuitive to me immediately why we needed to filter

Side note, in those cases, is it possible for us to see if there was a CRC written and use that for simple crc write?

This can be a P2+ but maybe we should track it somewhere (i.e. create a github issue?)

This could be a good optimization, filed #4177 for track

This should all just be a part of the rebase process. When we rebase, we need to list + read all the JSONs and do various conflict checkpoing. We might as well list + read the latest CRC file, too.

allisonport-db · 2025-02-19T21:03:58Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/checksum/CRCInfo.java

+    checkArgument(!batch.getColumnVector(getSchemaIndex(TABLE_SIZE_BYTES)).isNullAt(rowId));
+    checkArgument(!batch.getColumnVector(getSchemaIndex(NUM_FILES)).isNullAt(rowId));


I think technically we want to do the isNullAt check before we access the value. Basically we should prefix any CV access with a null check

Ah I found the util function, you should just use InternalUtils.requireNonNull for all the required fields

See Metadata.fromColumnVector for examples on how to use it

kernel/kernel-api/src/test/scala/io/delta/kernel/internal/checksum/ChecksumWriterSuite.scala

allisonport-db · 2025-02-19T23:01:38Z

kernel/kernel-api/src/test/scala/io/delta/kernel/internal/checksum/ChecksumWriterSuite.scala

+    )
+  }
+
+  private def checkMetadata(expectedMetadata: Metadata, actualMetadataRow: Row): Unit = {


As long as we fix these checkMetadata and checkProtocol methods in another PR I am good with this for now

allisonport-db · 2025-02-19T23:30:19Z

.../kernel-defaults/src/test/scala/io/delta/kernel/defaults/ChecksumSimpleComparisonSuite.scala

+    }
+  }
+
+  /**


I'm a little concerned that this method of testing isn't going to scale well, especially as we add support for other operations besides append for example.

Is there maybe an easier way we can do this or maybe this is overkill?

How do we test CRC writes within delta-spark? Can we match that level of testing?

Off the top of my head I think I would (1) want to test the crc content is correct and (2) check that delta spark can read/use the crc generated by Kernel. I'm not sure we need to assert they are exactly the same.

Open to differing opinions but would like to avoid overcomplicating our tests if possible.

Otherwise maybe there's at least some way to simplify these tests? For example, I think we can insert values into spark using the DF writer instead of building a SQL query. I also wonder is there any major difference for partitioned vs unpartitioned for CRC specifically?

Also if we need this, I think maybe there are simpler ways to do the conversion from Spark table to Delta actions. Can you try using TableImpl.getRawChanges maybe? You can possibly just query the file actions and then directly commit them to the Kernel table.

I'm thinking it would be much simpler and better to basically

In the test, make the changes directly to the Spark table (self defining)

Call fx to convert to Kernel commit (this doesn't require things like partition info, as that can be gathered from the add files I think. it should just take the version to convert as input)

Check CRC is the same

Spark does the a/b comparison, (not at file level though)

delta/spark/src/test/scala/org/apache/spark/sql/delta/ChecksumSuite.scala

Line 127 in 3e3c01b

assert(checksum.copy(txnId = None) === computedChecksum)

I think the value of this test is to ensure our behavior(especially file size collection) is the same as spark

Can you try using TableImpl.getRawChanges maybe?
This is an excellent suggestion! It simply the code a lot

aae9d64 is what I was trying to outline earlier but hard to explain

...el/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableWriteWithCrcSuite.scala

allisonport-db · 2025-02-19T23:51:28Z

...el/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableWriteWithCrcSuite.scala

+    assert(columnarBatches.hasNext, "Empty checksum file")
+    val crcRow = columnarBatches.next()
+    assert(crcRow.getSize === 1, s"Expected single row, found ${crcRow.getSize}")
+


How did you decide what to test content wise? Can you maybe explain what we check vs don't check and why?

allisonport-db · 2025-02-20T19:01:32Z

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableWriteSuiteBase.scala

@@ -460,4 +462,11 @@ trait DeltaTableWriteSuiteBase extends AnyFunSuite with TestUtils {
        .stream()
        .anyMatch(hook => hook.getType == PostCommitHookType.CHECKPOINT) === isReadyForCheckpoint)
  }
+


please document that all child suites should use this instead of .commit directly, and explain why this is!! also maybe put it at the top of the file in all caps (in DeltaTableWriteSuite)

allisonport-db · 2025-02-20T19:03:27Z

...el/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableWriteWithCrcSuite.scala

+    checkChecksum(path, expSchema)
+  }
+
+  def checkChecksum(


please document what this does and doesn't check

Update the method name to verifyChecksumValid and add docs

allisonport-db · 2025-02-20T19:03:55Z

...el/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableWriteWithCrcSuite.scala

+import io.delta.kernel.types.StructType
+import io.delta.kernel.utils.{CloseableIterable, FileStatus}
+
+class DeltaTableWriteWithCrcSuite extends DeltaTableWritesSuite {


class docs might also be useful here (i.e. we just check that crc is generated for all txn basically right? nothing more?)

vkorukanti

LGTM, some followup test work which can be in followup PRs.

vkorukanti · 2025-02-20T19:07:04Z

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableWritesSuite.scala

@@ -794,6 +797,8 @@ class DeltaTableWritesSuite extends DeltaTableWriteSuiteBase with ParquetSuiteBa

  test("insert into table - idempotent writes") {
    withTempDirAndEngine { (tblPath, engine) =>
+      // TODO: re-enable when CRC_FULL post commit hook is added, txn2 requires CRC_FULL
+      assume(this.suiteName != ("DeltaTableWriteWithCrcSuite"))


can we add this to the ingore list in the DeltaTableWriteWithCrcSuite? This can be a separate PR. I think we may need support for ingoreList similar to delta-spark tests.

we can make IgnoreTests a mixin trait

Will address it seperately

vkorukanti · 2025-02-20T19:08:49Z

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableWriteSuiteBase.scala

@@ -460,4 +462,11 @@ trait DeltaTableWriteSuiteBase extends AnyFunSuite with TestUtils {
        .stream()
        .anyMatch(hook => hook.getType == PostCommitHookType.CHECKPOINT) === isReadyForCheckpoint)
  }
+
+  protected def commitTransaction(


Please add an oneliner doc to this base class on the need to commit through this method.

allisonport-db

LGTM

allisonport-db · 2025-02-20T23:27:28Z

.../kernel-defaults/src/test/scala/io/delta/kernel/defaults/ChecksumSimpleComparisonSuite.scala

+    assert(crc1.getMetadata.getPartitionColNames === crc2.getMetadata.getPartitionColNames)
+  }
+
+  // TODO docs


can you add these docs in one of the follow up PRs?

huan233usc added 9 commits February 6, 2025 17:43

basic crc simple writer

1c05213

first test

ca06a90

add tests

3caed70

add tests

3b7dc2c

fix scala format

2f58dfa

add unit test

e7f4beb

add doc

52bda11

format java

d1626dc

format scala

2a1092d

huan233usc self-assigned this Feb 10, 2025

huan233usc added 5 commits February 10, 2025 12:48

fix test

6c8069f

fix test

471e172

refactor 1

8c3c33b

refactor 2

5d0b5a7

remove redundant doc

6796fa5

huan233usc commented Feb 10, 2025

View reviewed changes

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableWritesSuite.scala Show resolved Hide resolved

huan233usc added 2 commits February 10, 2025 15:10

fix doc

2a93c49

fix scala

826f623

huan233usc marked this pull request as ready for review February 10, 2025 23:18

huan233usc requested review from vkorukanti, scottsand-db and allisonport-db February 10, 2025 23:18

scottsand-db requested changes Feb 12, 2025

View reviewed changes

scottsand-db reviewed Feb 12, 2025

View reviewed changes

kernel/kernel-api/src/test/scala/io/delta/kernel/internal/checksum/ChecksumWriterSuite.scala Outdated Show resolved Hide resolved

scottsand-db reviewed Feb 12, 2025

View reviewed changes

huan233usc added 5 commits February 12, 2025 14:19

resolve comment

ebaf2c8

fix indent

000a36d

fix test

0c425be

fix indent

e972c31

fix indent

73df627

huan233usc added 3 commits February 19, 2025 12:34

typo

e7710cb

grouping

0bec898

todo

d392927

huan233usc requested a review from scottsand-db February 19, 2025 20:39

allisonport-db reviewed Feb 19, 2025

View reviewed changes

huan233usc mentioned this pull request Feb 19, 2025

[Feature Request]On simple crc generation, check if the (commit version - 1)'s crc file present. #4177

Open

8 tasks

huan233usc added 2 commits February 19, 2025 14:11

fix doc

f60b9c6

fix check

c4ae5e5

huan233usc requested a review from allisonport-db February 19, 2025 22:28

allisonport-db reviewed Feb 20, 2025

View reviewed changes

huan233usc added 4 commits February 19, 2025 16:34

fix test

a6be52b

fix test

eb1bca1

fix test

941d9c8

utils method

a14d42e

huan233usc requested a review from allisonport-db February 20, 2025 01:59

huan233usc added 3 commits February 19, 2025 19:01

fix scala

c4462a7

fix test

744beaf

revert

562cfe4

scottsand-db approved these changes Feb 20, 2025

View reviewed changes

allisonport-db reviewed Feb 20, 2025

View reviewed changes

vkorukanti approved these changes Feb 20, 2025

View reviewed changes

huan233usc added 3 commits February 20, 2025 11:53

add docs

b639110

patch delta-io@aae9d64

5a34d7e

remove patch file

b9b4aac

huan233usc requested a review from allisonport-db February 20, 2025 21:08

huan233usc added 2 commits February 20, 2025 14:07

Merge branch 'master' into crc_simple

7900389

resolve commit

a2e2883

allisonport-db approved these changes Feb 20, 2025

View reviewed changes

scottsand-db merged commit 7e85686 into delta-io:master Feb 20, 2025
21 checks passed

allisonport-db reviewed Feb 20, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Kernel]Add simple crc post commit for incremental crc writing. #4134

[Kernel]Add simple crc post commit for incremental crc writing. #4134

huan233usc commented Feb 8, 2025 •

edited

Loading

scottsand-db left a comment

scottsand-db Feb 12, 2025

scottsand-db Feb 12, 2025

huan233usc Feb 12, 2025

huan233usc Feb 14, 2025

scottsand-db Feb 12, 2025

huan233usc Feb 12, 2025

huan233usc Feb 14, 2025

allisonport-db Feb 19, 2025

allisonport-db Feb 19, 2025

huan233usc Feb 19, 2025

scottsand-db Feb 20, 2025

allisonport-db Feb 19, 2025

allisonport-db Feb 19, 2025

allisonport-db Feb 19, 2025

allisonport-db Feb 19, 2025

allisonport-db Feb 19, 2025

allisonport-db Feb 19, 2025

huan233usc Feb 20, 2025

allisonport-db Feb 20, 2025

allisonport-db Feb 19, 2025

allisonport-db Feb 20, 2025

allisonport-db Feb 20, 2025

huan233usc Feb 20, 2025

allisonport-db Feb 20, 2025

vkorukanti left a comment

vkorukanti Feb 20, 2025

allisonport-db Feb 20, 2025

huan233usc Feb 20, 2025

vkorukanti Feb 20, 2025

allisonport-db left a comment

allisonport-db Feb 20, 2025

		checkArgument(!batch.getColumnVector(getSchemaIndex(TABLE_SIZE_BYTES)).isNullAt(rowId));
		checkArgument(!batch.getColumnVector(getSchemaIndex(NUM_FILES)).isNullAt(rowId));

+                  }
+                }
+                /**

[Kernel]Add simple crc post commit for incremental crc writing. #4134

[Kernel]Add simple crc post commit for incremental crc writing. #4134

Conversation

huan233usc commented Feb 8, 2025 • edited Loading

Which Delta project/connector is this regarding?

Description

How was this patch tested?

Does this PR introduce any user-facing changes?

scottsand-db left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vkorukanti left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

allisonport-db left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

huan233usc commented Feb 8, 2025 •

edited

Loading