update schema with record of largest debezium ts_ms

Signed-off-by: zenghua <[email protected]>
lakesoul-io · Aug 1, 2024 · ff9b3c1 · ff9b3c1
1 parent 26e43f8
commit ff9b3c1
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 11 deletions.
diff --git a/...k/src/main/java/org/apache/flink/lakesoul/sink/committer/LakeSoulSinkGlobalCommitter.java b/...k/src/main/java/org/apache/flink/lakesoul/sink/committer/LakeSoulSinkGlobalCommitter.java
@@ -230,6 +230,7 @@ public List<LakeSoulMultiTableSinkGlobalCommittable> commit(
                                 sparkSchema,
                                 partitionKeys.rangeKeys,
                                 partitionKeys.primaryKeys);
+                LOG.info("{}", equalOrCanCastTuple3);
                 String equalOrCanCast = equalOrCanCastTuple3._1();
                 boolean schemaChanged = (boolean) equalOrCanCastTuple3._2();
                 StructType mergeStructType = equalOrCanCastTuple3._3();
@@ -273,12 +274,10 @@ public List<LakeSoulMultiTableSinkGlobalCommittable> commit(
                     if (equalOrCanCast.contains("Change of Partition Column") || equalOrCanCast.contains("Change of Primary Key Column")) {
                         throw new IOException(equalOrCanCast);
                     }
-//                    for (LakeSoulMultiTableSinkCommittable committable : committableList) {
-//                        if (committable.getTsMs() > schemaLastChangeTime) {
-//                            LOG.error("incompatible cast data created and delayThreshold time: {}, dml create time: {}", schemaLastChangeTime, committable.getTsMs());
-//                            throw new IOException(equalOrCanCast);
-//                        }
-//                    }
+                    if (entry.getValue().f0 > schemaLastChangeTime) {
+                        LOG.error("incompatible cast data created and delayThreshold time: {}, dml create time: {}", schemaLastChangeTime, entry.getValue().f0);
+                        throw new IOException(equalOrCanCast);
+                    }
                 }
             }
         }

diff --git a/lakesoul-spark/src/test/scala/org/apache/spark/sql/lakesoul/DDLSuite.scala b/lakesoul-spark/src/test/scala/org/apache/spark/sql/lakesoul/DDLSuite.scala
@@ -486,9 +486,9 @@ abstract class DDLTestBase extends QueryTest with SQLTestUtils {
     }
   }
 
-  test("read") {
-    sql("show create table TestBinarySourceRecordSink").show(false)
-    sql("desc TestBinarySourceRecordSink").show()
-    sql("select * from TestBinarySourceRecordSink").show()
-  }
+  //  test("read") {
+  //    sql("show create table TestBinarySourceRecordSink").show(false)
+  //    sql("desc TestBinarySourceRecordSink").show()
+  //    sql("select * from TestBinarySourceRecordSink").show()
+  //  }
 }