This is an automated email from the ASF dual-hosted git repository.
morrySnow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 44aaa1e87eb [fix](binlog) Align row binlog after schema with before
values (#65438)
44aaa1e87eb is described below
commit 44aaa1e87ebae9bbd8b582f97a6b4d30d075fd58
Author: yujun <[email protected]>
AuthorDate: Fri Jul 10 16:27:48 2026 +0800
[fix](binlog) Align row binlog after schema with before values (#65438)
### What problem does this PR solve?
Problem Summary:
Row binlog `MIN_DELTA` does not output separate schemas for
update-before rows and update-after rows. Instead, it materializes both
row kinds through one unified output schema, which is the generated
after/output value schema.
That means an update-before row also has to be written into the after
value slots. For MoW row binlog queries with historical values enabled,
the before values come from the nullable `__DORIS_BEFORE_*` mirror
columns. Before this change, FE generated after value columns by copying
the user column definition directly. When the user column was a complex
`NOT NULL` type such as `BITMAP`, the after/output slot stayed
non-nullable, but `MIN_DELTA` still had to write the nullable before
value into that same slot.
This schema mismatch could crash BE during `MIN_DELTA` row
materialization with a bad cast from `ColumnNullable` to
`ColumnComplexType<BITMAP>`. The fix is to make generated after value
columns follow the same nullable and default-cleared contract as before
value columns, so the unified `MIN_DELTA` output schema can safely carry
both before rows and after rows.
This PR also adds FE unit coverage for row binlog schema generation and
a normal row binlog regression case that reproduces the crash with a
`BITMAP NOT NULL` column on `MIN_DELTA`.
---
.../main/java/org/apache/doris/catalog/Column.java | 5 +
.../catalog/OlapTableRowBinlogSchemaTest.java | 30 ++-
.../row_binlog_p0/test_binlog_changes_syntax.out | 212 ++++++++++++++++++
.../test_binlog_changes_syntax.groovy | 236 +++++++--------------
4 files changed, 320 insertions(+), 163 deletions(-)
diff --git a/fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java
b/fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java
index f3acb582001..923d1954d77 100644
--- a/fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java
+++ b/fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java
@@ -101,6 +101,11 @@ public class Column implements GsonPostProcessable {
Column afterValueColumn = new Column(column);
afterValueColumn.setComment("after value (" + column.getName() + ")");
afterValueColumn.setAggregationType(AggregateType.NONE, true);
+ afterValueColumn.setIsAllowNull(true);
+ // clear default value
+ afterValueColumn.defaultValue = null;
+ afterValueColumn.defaultValueExprDef = null;
+ afterValueColumn.realDefaultValue = null;
return afterValueColumn;
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableRowBinlogSchemaTest.java
b/fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableRowBinlogSchemaTest.java
index f6130782d73..972f0d82267 100755
---
a/fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableRowBinlogSchemaTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableRowBinlogSchemaTest.java
@@ -33,7 +33,7 @@ public class OlapTableRowBinlogSchemaTest {
long baseIndexId = 1L;
Column key = new Column("k1", PrimitiveType.INT);
key.setIsKey(true);
- Column value = new Column("v1", PrimitiveType.INT);
+ Column value = new Column("v1", Type.INT, false, null, false, "7", "");
value.setIsKey(false);
List<Column> baseSchema = Lists.newArrayList(key, value);
@@ -62,6 +62,8 @@ public class OlapTableRowBinlogSchemaTest {
public void testRowBinlogSchemaOnEnable() {
OlapTable tableWithoutBefore =
newTestTable(BinlogTestUtils.newTestRowBinlogConfig(true, false));
Assertions.assertTrue(tableWithoutBefore.needRowBinlog());
+
Assertions.assertFalse(tableWithoutBefore.getBaseSchema(true).get(1).isAllowNull());
+ Assertions.assertEquals("7",
tableWithoutBefore.getBaseSchema(true).get(1).getDefaultValue());
List<String> tableWithoutBeforeColumns =
tableWithoutBefore.getRowBinlogMeta().getSchema(true).stream().map(Column::getName)
.collect(Collectors.toList());
@@ -70,17 +72,39 @@ public class OlapTableRowBinlogSchemaTest {
Assertions.assertEquals(tableWithoutBeforeColumns.indexOf(Column.BINLOG_OPERATION_COL),
3);
Assertions.assertEquals(tableWithoutBeforeColumns.indexOf(Column.BINLOG_TIMESTAMP_COL),
4);
Assertions.assertEquals(tableWithoutBeforeColumns.size(), 5);
+ Column afterColumnWithoutBefore =
tableWithoutBefore.getRowBinlogMeta().getSchema(true).stream()
+ .filter(column -> column.getName().equals("v1"))
+ .findFirst()
+ .orElseThrow();
+ Assertions.assertTrue(afterColumnWithoutBefore.isAllowNull());
+ Assertions.assertNull(afterColumnWithoutBefore.getDefaultValue());
+
Assertions.assertNull(afterColumnWithoutBefore.getDefaultValueExprDef());
+ Assertions.assertNull(afterColumnWithoutBefore.getRealDefaultValue());
OlapTable tableWithBefore =
newTestTable(BinlogTestUtils.newTestRowBinlogConfig(true, true));
Assertions.assertTrue(tableWithBefore.needRowBinlog());
+ List<Column> rowBinlogSchemaWithBefore =
tableWithBefore.getRowBinlogMeta().getSchema(true);
List<String> tableWithBeforeColumns =
-
tableWithBefore.getRowBinlogMeta().getSchema(true).stream().map(Column::getName)
- .collect(Collectors.toList());
+
rowBinlogSchemaWithBefore.stream().map(Column::getName).collect(Collectors.toList());
Assertions.assertTrue(tableWithBeforeColumns.contains(Column.generateBeforeColName("v1")));
Assertions.assertEquals(tableWithBeforeColumns.indexOf(Column.BINLOG_LSN_COL),
3);
Assertions.assertEquals(tableWithBeforeColumns.indexOf(Column.BINLOG_OPERATION_COL),
4);
Assertions.assertEquals(tableWithBeforeColumns.indexOf(Column.BINLOG_TIMESTAMP_COL),
5);
Assertions.assertEquals(tableWithBeforeColumns.size(), 6);
+ Column afterColumnWithBefore = rowBinlogSchemaWithBefore.stream()
+ .filter(column -> column.getName().equals("v1"))
+ .findFirst()
+ .orElseThrow();
+ Assertions.assertTrue(afterColumnWithBefore.isAllowNull());
+ Assertions.assertNull(afterColumnWithBefore.getDefaultValue());
+ Column beforeColumn = rowBinlogSchemaWithBefore.stream()
+ .filter(column ->
column.getName().equals(Column.generateBeforeColName("v1")))
+ .findFirst()
+ .orElseThrow();
+ Assertions.assertTrue(beforeColumn.isAllowNull());
+ Assertions.assertNull(beforeColumn.getDefaultValue());
+ Assertions.assertNull(beforeColumn.getDefaultValueExprDef());
+ Assertions.assertNull(beforeColumn.getRealDefaultValue());
}
@Test
diff --git a/regression-test/data/row_binlog_p0/test_binlog_changes_syntax.out
b/regression-test/data/row_binlog_p0/test_binlog_changes_syntax.out
new file mode 100644
index 00000000000..f2fa93f1918
--- /dev/null
+++ b/regression-test/data/row_binlog_p0/test_binlog_changes_syntax.out
@@ -0,0 +1,212 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !dup_append_range --
+3 30 c y 0
+3 31 c2 \N 0
+4 40 \N z 0
+
+-- !dup_detail_range --
+3 30 c y 0
+3 31 c2 \N 0
+4 40 \N z 0
+
+-- !dup_min_delta_range --
+3 30 c y 0
+3 31 c2 \N 0
+4 40 \N z 0
+
+-- !dup_detail_start --
+3 30 0
+3 31 0
+4 40 0
+5 50 0
+
+-- !dup_detail_end --
+1 10 0
+2 20 0
+3 30 0
+3 31 0
+4 40 0
+
+-- !dup_detail_all --
+1 10 0
+2 20 0
+3 30 0
+3 31 0
+4 40 0
+5 50 0
+
+-- !dup_incr_empty --
+1 10 0
+2 20 0
+3 30 0
+3 31 0
+4 40 0
+5 50 0
+
+-- !dup_reversed --
+
+-- !dup_far_future --
+
+-- !dup_full_cover --
+1 10 0
+2 20 0
+3 30 0
+3 31 0
+4 40 0
+5 50 0
+
+-- !dup_binlog_tvf --
+1 10 0
+2 20 0
+3 30 0
+3 31 0
+4 40 0
+5 50 0
+
+-- !mow_append_range --
+2 20 b 0
+4 40 d 0
+5 50 e 0
+
+-- !mow_min_delta_range --
+1 10 a 2
+1 13 a3 3
+2 21 b1 0
+3 30 c 1
+4 40 d 0
+
+-- !mow_detail_range --
+1 10 a 2
+1 11 a1 2
+1 11 a1 3
+1 12 a2 2
+1 12 a2 3
+1 13 a3 3
+2 \N \N 2
+2 20 b 0
+2 20 b 1
+2 21 b1 3
+3 30 c 1
+4 40 d 0
+5 50 e 0
+5 50 e 1
+
+-- !mow_detail_start --
+1 10 2
+1 11 2
+1 11 3
+1 12 2
+1 12 3
+1 13 3
+2 \N 2
+2 20 0
+2 20 1
+2 21 3
+3 30 1
+4 40 0
+5 50 0
+5 50 1
+6 60 0
+
+-- !mow_detail_end --
+1 10 0
+1 10 2
+1 11 2
+1 11 3
+1 12 2
+1 12 3
+1 13 3
+2 \N 2
+2 20 0
+2 20 1
+2 21 3
+3 30 0
+3 30 1
+4 40 0
+5 50 0
+5 50 1
+
+-- !mow_detail_all --
+1 10 0
+1 10 2
+1 11 2
+1 11 3
+1 12 2
+1 12 3
+1 13 3
+2 \N 2
+2 20 0
+2 20 1
+2 21 3
+3 30 0
+3 30 1
+4 40 0
+5 50 0
+5 50 1
+6 60 0
+
+-- !mow_min_delta_all --
+1 13 0
+2 21 0
+4 40 0
+6 60 0
+
+-- !mow_incr_empty --
+1 13 0
+2 21 0
+4 40 0
+6 60 0
+
+-- !mow_reversed --
+
+-- !mow_full_cover --
+1 10 0
+1 10 2
+1 11 2
+1 11 3
+1 12 2
+1 12 3
+1 13 3
+2 \N 2
+2 20 0
+2 20 1
+2 21 3
+3 30 0
+3 30 1
+4 40 0
+5 50 0
+5 50 1
+6 60 0
+
+-- !seq_detail --
+1 100 2
+1 300 3
+
+-- !seq_min_delta --
+1 100 2
+1 300 3
+
+-- !seq_visible --
+300
+
+-- !part_detail --
+1 10 a 1000 2
+1 11 a 1000 2
+1 11 a 1000 3
+1 11 b 1000 2
+1 11 b 1000 3
+1 11 b 2000 3
+
+-- !part_min_delta --
+1 10 a 1000 2
+1 11 b 2000 3
+
+-- !part_visible --
+11 b 2000
+
+-- !part_append --
+
+-- !bitmap_min_delta --
+1 1,2 2
+1 3,4 3
+
diff --git
a/regression-test/suites/row_binlog_p0/test_binlog_changes_syntax.groovy
b/regression-test/suites/row_binlog_p0/test_binlog_changes_syntax.groovy
index b2212177cc8..e9e2f758cec 100644
--- a/regression-test/suites/row_binlog_p0/test_binlog_changes_syntax.groovy
+++ b/regression-test/suites/row_binlog_p0/test_binlog_changes_syntax.groovy
@@ -29,16 +29,15 @@ suite("test_binlog_changes_syntax", "nonConcurrent") {
def mowTable = "changes_mow"
def mowSeqTable = "changes_mow_seq"
def mowPartialTable = "changes_mow_partial"
+ def mowBitmapTable = "changes_mow_bitmap"
def incrTimeFormat = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
- // Helper: stable LSN sort accessor.
- def asStr = { row, idx -> row[idx] == null ? "null" : row[idx].toString() }
-
try {
sql "DROP TABLE IF EXISTS ${dupTable}"
sql "DROP TABLE IF EXISTS ${mowTable}"
sql "DROP TABLE IF EXISTS ${mowSeqTable}"
sql "DROP TABLE IF EXISTS ${mowPartialTable}"
+ sql "DROP TABLE IF EXISTS ${mowBitmapTable}"
// ============================================================
// 1. DUPLICATE table — multi-column + nullable + multi-rowset writes.
@@ -91,128 +90,97 @@ suite("test_binlog_changes_syntax", "nonConcurrent") {
// 1.1 APPEND_ONLY [dupT0, dupT1]: 3 in-window inserts including the
// re-inserted key=3.
- def dupAppendRange = sql """
+ order_qt_dup_append_range """
SELECT id, v1, v2, v3, __DORIS_BINLOG_OP__
FROM ${dupTable}@incr('startTimestamp' = '${dupT0}',
"endTimestamp" = "${dupT1}",
"incrementType" = "APPEND_ONLY")
ORDER BY __DORIS_BINLOG_LSN__
"""
- assertEquals(3, dupAppendRange.size())
- assertEquals("3", asStr(dupAppendRange[0], 0))
- assertEquals("30", asStr(dupAppendRange[0], 1))
- assertEquals("c", asStr(dupAppendRange[0], 2))
- assertEquals("y", asStr(dupAppendRange[0], 3))
- assertEquals("0", asStr(dupAppendRange[0], 4))
- assertEquals("4", asStr(dupAppendRange[1], 0))
- assertEquals("null", asStr(dupAppendRange[1], 2))
- assertEquals("z", asStr(dupAppendRange[1], 3))
- assertEquals("3", asStr(dupAppendRange[2], 0))
- assertEquals("31", asStr(dupAppendRange[2], 1))
- assertEquals("c2", asStr(dupAppendRange[2], 2))
- assertEquals("null", asStr(dupAppendRange[2], 3))
// 1.2 DETAIL [dupT0, dupT1] must be identical to APPEND_ONLY for dup.
- def dupDetailRange = sql """
+ order_qt_dup_detail_range """
SELECT id, v1, v2, v3, __DORIS_BINLOG_OP__
FROM ${dupTable}@incr('startTimestamp' = '${dupT0}',
"endTimestamp" = "${dupT1}",
"incrementType" = "DETAIL")
ORDER BY __DORIS_BINLOG_LSN__
"""
- assertEquals(dupAppendRange, dupDetailRange)
// 1.3 MIN_DELTA on dup table should be equivalent to APPEND_ONLY:
// dup table has no DELETE / UPDATE in binlog, so per-key folding
// degenerates to "all inserts kept as APPEND".
- def dupMinDeltaRange = sql """
+ order_qt_dup_min_delta_range """
SELECT id, v1, v2, v3, __DORIS_BINLOG_OP__
FROM ${dupTable}@incr('startTimestamp' = '${dupT0}',
"endTimestamp" = "${dupT1}",
"incrementType" = "MIN_DELTA")
ORDER BY __DORIS_BINLOG_LSN__
"""
- assertEquals(dupAppendRange, dupMinDeltaRange)
// 1.4 startTimestamp only: covers in-window + late insert.
- def dupDetailStart = sql """
+ order_qt_dup_detail_start """
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${dupTable}@incr('startTimestamp' = '${dupT0}',
"incrementType" = "DETAIL")
ORDER BY __DORIS_BINLOG_LSN__
"""
- assertEquals(4, dupDetailStart.size())
- assertEquals("5", asStr(dupDetailStart[3], 0))
- assertEquals("50", asStr(dupDetailStart[3], 1))
// 1.5 endTimestamp only: covers seed + in-window inserts.
- def dupDetailEnd = sql """
+ order_qt_dup_detail_end """
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${dupTable}@incr("endTimestamp" = "${dupT1}",
"incrementType" = "DETAIL")
ORDER BY __DORIS_BINLOG_LSN__
"""
- assertEquals(5, dupDetailEnd.size())
- assertEquals("1", asStr(dupDetailEnd[0], 0))
- assertEquals("2", asStr(dupDetailEnd[1], 0))
// 1.6 No timestamps: equals @incr() and equals DETAIL of full binlog.
- def dupDetailAll = sql """
+ order_qt_dup_detail_all """
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${dupTable}@incr("incrementType" = "DETAIL")
ORDER BY __DORIS_BINLOG_LSN__
"""
- def dupIncrEmpty = sql """
+ order_qt_dup_incr_empty """
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${dupTable}@incr()
ORDER BY __DORIS_BINLOG_LSN__
"""
- assertEquals(6, dupDetailAll.size())
- assertEquals(dupDetailAll, dupIncrEmpty)
// 1.7 Degenerate windows.
// - start > end: should yield empty.
- def dupReversed = sql """
+ order_qt_dup_reversed """
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${dupTable}@incr('startTimestamp' = '${dupT1}',
"endTimestamp" = "${dupT0}",
"incrementType" = "DETAIL")
ORDER BY __DORIS_BINLOG_LSN__
"""
- assertEquals(0, dupReversed.size())
// - far-future start: empty.
- def dupFarFuture = sql """
+ order_qt_dup_far_future """
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${dupTable}@incr('startTimestamp' = '2999-01-01 00:00:00',
"incrementType" = "DETAIL")
ORDER BY __DORIS_BINLOG_LSN__
"""
- assertEquals(0, dupFarFuture.size())
// - far-past start + far-future end: equivalent to full binlog.
- def dupFullCover = sql """
+ order_qt_dup_full_cover """
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${dupTable}@incr('startTimestamp' = '1971-01-01 00:00:00',
"endTimestamp" = "2999-01-01 00:00:00",
"incrementType" = "DETAIL")
ORDER BY __DORIS_BINLOG_LSN__
"""
- assertEquals(dupDetailAll, dupFullCover)
// 1.8 Cross-check against binlog() TVF — DETAIL with full window must
// return the same rowset as the underlying binlog TVF for a dup
// table (since both expose only INSERT rows).
- def dupBinlogTvf = sql """
+ order_qt_dup_binlog_tvf """
SELECT id, v1, __DORIS_BINLOG_OP__
FROM binlog("table" = "${dupTable}")
ORDER BY __DORIS_BINLOG_LSN__
"""
- assertEquals(dupBinlogTvf.size(), dupDetailAll.size())
- for (int i = 0; i < dupBinlogTvf.size(); i++) {
- assertEquals(asStr(dupBinlogTvf[i], 0), asStr(dupDetailAll[i], 0))
- assertEquals(asStr(dupBinlogTvf[i], 1), asStr(dupDetailAll[i], 1))
- }
// ============================================================
// 2. MoW table (UNIQUE KEY + enable_unique_key_merge_on_write
@@ -294,21 +262,16 @@ suite("test_binlog_changes_syntax", "nonConcurrent") {
// key=4 INSERT(40) is APPEND.
// key=5 INSERT(50) is APPEND (the later DELETE does not undo
// APPEND in APPEND_ONLY semantics).
- def mowAppendRange = sql """
+ order_qt_mow_append_range """
SELECT id, v1, v2, __DORIS_BINLOG_OP__
FROM ${mowTable}@incr('startTimestamp' = '${mowT0}',
"endTimestamp" = "${mowT1}",
"incrementType" = "APPEND_ONLY")
ORDER BY id, __DORIS_BINLOG_LSN__
"""
- assertEquals(3, mowAppendRange.size())
- // Sort by id is enough to locate them.
- def appendIds = mowAppendRange.collect { asStr(it, 0) }.toSet()
- assertEquals(["2", "4", "5"] as Set, appendIds)
- mowAppendRange.each { assertEquals("0", asStr(it, 3)) }
// 2.2 MIN_DELTA [mowT0, mowT1]: per-key folded result.
- def mowMinDeltaRange = sql """
+ order_qt_mow_min_delta_range """
SELECT id, v1, v2, __DORIS_BINLOG_OP__
FROM ${mowTable}@incr('startTimestamp' = '${mowT0}',
"endTimestamp" = "${mowT1}",
@@ -324,38 +287,6 @@ suite("test_binlog_changes_syntax", "nonConcurrent") {
// (2, 21, 'b1', 0) -- key=2 did not exist before mowT0
// (3, 30, 'c', 1)
// (4, 40, 'd', 0)
- assertEquals(5, mowMinDeltaRange.size())
- def mdByKey = [:]
- mowMinDeltaRange.each {
- def k = asStr(it, 0)
- mdByKey.computeIfAbsent(k, { _ -> [] }).add(it)
- }
- // key=1
- assertEquals(2, mdByKey["1"].size())
- def k1Ops = mdByKey["1"].collect { asStr(it, 3) }.toSet()
- assertEquals(["2", "3"] as Set, k1Ops)
- def k1Before = mdByKey["1"].find { asStr(it, 3) == "2" }
- def k1After = mdByKey["1"].find { asStr(it, 3) == "3" }
- assertEquals("10", asStr(k1Before, 1))
- assertEquals("13", asStr(k1After, 1))
- assertEquals("a3", asStr(k1After, 2))
- // key=2: did not exist before mowT0; insert+delete+insert collapses to
- // a net APPEND(21).
- assertEquals(1, mdByKey["2"].size())
- assertEquals("0", asStr(mdByKey["2"][0], 3))
- assertEquals("21", asStr(mdByKey["2"][0], 1))
- assertEquals("b1", asStr(mdByKey["2"][0], 2))
- // key=3: pure DELETE
- assertEquals(1, mdByKey["3"].size())
- assertEquals("1", asStr(mdByKey["3"][0], 3))
- assertEquals("30", asStr(mdByKey["3"][0], 1))
- // key=4: pure APPEND
- assertEquals(1, mdByKey["4"].size())
- assertEquals("0", asStr(mdByKey["4"][0], 3))
- assertEquals("40", asStr(mdByKey["4"][0], 1))
- // key=5: insert-then-delete inside window collapses to nothing.
- assertEquals(null, mdByKey["5"])
-
// 2.3 DETAIL [mowT0, mowT1]: every raw binlog row in the window.
// Each "INSERT into UNIQUE KEY MoW that hits an existing key"
emits
// a UPDATE_BEFORE/UPDATE_AFTER pair instead of a plain INSERT.
After
@@ -369,53 +300,35 @@ suite("test_binlog_changes_syntax", "nonConcurrent") {
// key=4: INSERT(40) = 1 row
// key=5: INSERT(50) + DELETE(50) = 2 rows
// Total = 6 + 4 + 1 + 1 + 2 = 14.
- def mowDetailRange = sql """
+ order_qt_mow_detail_range """
SELECT id, v1, v2, __DORIS_BINLOG_OP__
FROM ${mowTable}@incr('startTimestamp' = '${mowT0}',
"endTimestamp" = "${mowT1}",
"incrementType" = "DETAIL")
ORDER BY __DORIS_BINLOG_LSN__
"""
- assertEquals(14, mowDetailRange.size())
- // Detail must contain every concrete value we ever wrote on key=1.
- def k1Vals = mowDetailRange.findAll { asStr(it, 0) == "1" }.collect {
asStr(it, 1) }.toSet()
- assertTrue(k1Vals.containsAll(["10", "11", "12", "13"] as Set))
// 2.4 startTimestamp only: includes the late (6,60).
- def mowDetailStart = sql """
+ order_qt_mow_detail_start """
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${mowTable}@incr('startTimestamp' = '${mowT0}',
"incrementType" = "DETAIL")
ORDER BY __DORIS_BINLOG_LSN__
"""
- assertEquals(15, mowDetailStart.size())
- assertEquals("6", asStr(mowDetailStart[14], 0))
- assertEquals("60", asStr(mowDetailStart[14], 1))
- assertEquals("0", asStr(mowDetailStart[14], 2))
// 2.5 endTimestamp only: includes the seed (1,10) and (3,30).
- def mowDetailEnd = sql """
+ order_qt_mow_detail_end """
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${mowTable}@incr("endTimestamp" = "${mowT1}",
"incrementType" = "DETAIL")
ORDER BY __DORIS_BINLOG_LSN__
"""
- assertEquals(16, mowDetailEnd.size())
- assertEquals("1", asStr(mowDetailEnd[0], 0))
- assertEquals("10", asStr(mowDetailEnd[0], 1))
- assertEquals("0", asStr(mowDetailEnd[0], 2))
- assertEquals("3", asStr(mowDetailEnd[1], 0))
- assertEquals("30", asStr(mowDetailEnd[1], 1))
- assertEquals("0", asStr(mowDetailEnd[1], 2))
- def mowDetailAll = sql """
+ order_qt_mow_detail_all """
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${mowTable}@incr("incrementType" = "DETAIL")
ORDER BY __DORIS_BINLOG_LSN__
"""
- // 2 seed + 14 in-window + 1 late = 17.
- assertEquals(17, mowDetailAll.size())
-
// 2.6 MIN_DELTA across full binlog: per-key folding from binlog start
// (every key starts non-existent) to the final visible state.
// Even the "seed" inserts before mowT0 are inside the binlog
@@ -427,57 +340,35 @@ suite("test_binlog_changes_syntax", "nonConcurrent") {
// key=4: APPEND(40)
// key=5: SKIP
// key=6: APPEND(60)
- def mowMinDeltaAll = sql """
+ order_qt_mow_min_delta_all """
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${mowTable}@incr("incrementType" = "MIN_DELTA")
ORDER BY id, __DORIS_BINLOG_OP__, v1
"""
- assertEquals(4, mowMinDeltaAll.size())
- def mdAllByKey = [:]
- mowMinDeltaAll.each {
- mdAllByKey.computeIfAbsent(asStr(it, 0), { _ -> [] }).add(it)
- }
- assertEquals(1, mdAllByKey["1"].size())
- assertEquals("0", asStr(mdAllByKey["1"][0], 2))
- assertEquals("13", asStr(mdAllByKey["1"][0], 1))
- // key=2 over the full timeline: the key did not exist before mowT0,
- // so the net effect is APPEND(21). MIN_DELTA op = 0.
- assertEquals(1, mdAllByKey["2"].size())
- assertEquals("0", asStr(mdAllByKey["2"][0], 2))
- assertEquals("21", asStr(mdAllByKey["2"][0], 1))
- // key=3: full-timeline insert(30) + delete folds to SKIP.
- assertEquals(null, mdAllByKey["3"])
- assertEquals("0", asStr(mdAllByKey["4"][0], 2))
- assertEquals(null, mdAllByKey["5"])
- assertEquals("0", asStr(mdAllByKey["6"][0], 2))
- assertEquals("60", asStr(mdAllByKey["6"][0], 1))
// 2.7 Empty timestamps: min-delta binlog and equivalence with @incr().
- def mowIncrEmpty = sql """
+ order_qt_mow_incr_empty """
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${mowTable}@incr()
ORDER BY __DORIS_BINLOG_LSN__
"""
- assertEquals(mowMinDeltaAll, mowIncrEmpty)
// 2.8 Degenerate window: start > end -> empty.
- def mowReversed = sql """
+ order_qt_mow_reversed """
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${mowTable}@incr('startTimestamp' = '${mowT1}',
"endTimestamp" = "${mowT0}",
"incrementType" = "MIN_DELTA")
"""
- assertEquals(0, mowReversed.size())
// 2.9 Far-past + far-future window covers everything.
- def mowFullCover = sql """
+ order_qt_mow_full_cover """
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${mowTable}@incr('startTimestamp' = '1971-01-01 00:00:00',
"endTimestamp" = "2999-01-01 00:00:00",
"incrementType" = "DETAIL")
ORDER BY __DORIS_BINLOG_LSN__
"""
- assertEquals(mowDetailAll, mowFullCover)
// ============================================================
// 3. MoW table with sequence column.
@@ -531,33 +422,25 @@ suite("test_binlog_changes_syntax", "nonConcurrent") {
// 3.1 DETAIL only captures physically-applied writes. The out-of-order
// write (seq=3) is rejected and never lands in binlog, so only the
// accepted update (seq=10) shows up.
- def seqDetail = sql """
+ order_qt_seq_detail """
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${mowSeqTable}@incr('startTimestamp' = '${seqT0}',
"endTimestamp" = "${seqT1}",
"incrementType" = "DETAIL")
ORDER BY __DORIS_BINLOG_LSN__
"""
- // Only the accepted update produces 1 BEFORE/AFTER pair = 2 rows.
- assertEquals(2, seqDetail.size())
// 3.2 MIN_DELTA collapses to BEFORE(100) -> AFTER(latest visible
value).
- def seqMinDelta = sql """
+ order_qt_seq_min_delta """
SELECT id, v1, __DORIS_BINLOG_OP__
FROM ${mowSeqTable}@incr('startTimestamp' = '${seqT0}',
"endTimestamp" = "${seqT1}",
"incrementType" = "MIN_DELTA")
ORDER BY __DORIS_BINLOG_OP__
"""
- assertEquals(2, seqMinDelta.size())
- // Op 2 is UPDATE_BEFORE, op 3 is UPDATE_AFTER.
- assertEquals("2", asStr(seqMinDelta[0], 2))
- assertEquals("100", asStr(seqMinDelta[0], 1))
- assertEquals("3", asStr(seqMinDelta[1], 2))
- // The visible AFTER value must be the latest visible row. Read it back
- // from the table to avoid hard-coding implementation details.
- def visible = sql "SELECT v1 FROM ${mowSeqTable} WHERE id = 1"
- assertEquals(asStr(visible[0], 0), asStr(seqMinDelta[1], 1))
+ order_qt_seq_visible """
+ SELECT v1 FROM ${mowSeqTable} WHERE id = 1
+ """
// ============================================================
// 4. MoW table with partial column update. Each partial update
@@ -597,44 +480,77 @@ suite("test_binlog_changes_syntax", "nonConcurrent") {
def partT1 = incrTimeFormat.format(new Date())
// 4.1 DETAIL: 3 partial updates -> 6 raw binlog rows.
- def partDetail = sql """
+ order_qt_part_detail """
SELECT id, v1, v2, v3, __DORIS_BINLOG_OP__
FROM ${mowPartialTable}@incr('startTimestamp' = '${partT0}',
"endTimestamp" = "${partT1}",
"incrementType" = "DETAIL")
ORDER BY __DORIS_BINLOG_LSN__
"""
- assertEquals(6, partDetail.size())
// 4.2 MIN_DELTA: collapses to BEFORE/AFTER pair for key=1, where AFTER
// reflects the merged final visible row.
- def partMinDelta = sql """
+ order_qt_part_min_delta """
SELECT id, v1, v2, v3, __DORIS_BINLOG_OP__
FROM ${mowPartialTable}@incr('startTimestamp' = '${partT0}',
"endTimestamp" = "${partT1}",
"incrementType" = "MIN_DELTA")
ORDER BY __DORIS_BINLOG_OP__
"""
- assertEquals(2, partMinDelta.size())
- assertEquals("2", asStr(partMinDelta[0], 4))
- assertEquals("10", asStr(partMinDelta[0], 1))
- assertEquals("a", asStr(partMinDelta[0], 2))
- assertEquals("1000", asStr(partMinDelta[0], 3))
- assertEquals("3", asStr(partMinDelta[1], 4))
- def partVisible = sql "SELECT v1, v2, v3 FROM ${mowPartialTable} WHERE
id = 1"
- assertEquals(asStr(partVisible[0], 0), asStr(partMinDelta[1], 1))
- assertEquals(asStr(partVisible[0], 1), asStr(partMinDelta[1], 2))
- assertEquals(asStr(partVisible[0], 2), asStr(partMinDelta[1], 3))
+ order_qt_part_visible """
+ SELECT v1, v2, v3 FROM ${mowPartialTable} WHERE id = 1
+ """
// 4.3 APPEND_ONLY in the same window: pure partial updates on an
// existing key produce no APPEND rows.
- def partAppend = sql """
+ order_qt_part_append """
SELECT id, __DORIS_BINLOG_OP__
FROM ${mowPartialTable}@incr('startTimestamp' = '${partT0}',
"endTimestamp" = "${partT1}",
"incrementType" = "APPEND_ONLY")
"""
- assertEquals(0, partAppend.size())
+
+ // ============================================================
+ // 5. MoW table with BITMAP NOT NULL.
+ // This specifically exercises MIN_DELTA UPDATE_BEFORE/AFTER on a
+ // complex-type value column. Before the Column.java fix, the
+ // UPDATE_BEFORE row read from nullable __BEFORE__b__ would be
+ // copied into the non-nullable after-slot and BE would crash with
+ // "Bad cast from type ColumnNullable to ColumnComplexType<BITMAP>".
+ // ============================================================
+ sql """
+ CREATE TABLE ${mowBitmapTable} (
+ id BIGINT,
+ b BITMAP NOT NULL
+ ) ENGINE=OLAP
+ UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "light_schema_change" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+ sql "INSERT INTO ${mowBitmapTable} VALUES (1,
BITMAP_FROM_STRING('1,2'))"
+ sql "sync"
+ sleep(1200)
+ def bitmapT0 = incrTimeFormat.format(new Date())
+ sleep(1200)
+ sql "INSERT INTO ${mowBitmapTable} VALUES (1,
BITMAP_FROM_STRING('3,4'))"
+ sql "sync"
+ sleep(1200)
+ def bitmapT1 = incrTimeFormat.format(new Date())
+
+ order_qt_bitmap_min_delta """
+ SELECT id, BITMAP_TO_STRING(b), __DORIS_BINLOG_OP__
+ FROM ${mowBitmapTable}@incr('startTimestamp' = '${bitmapT0}',
+ "endTimestamp" = "${bitmapT1}",
+ "incrementType" = "MIN_DELTA")
+ ORDER BY __DORIS_BINLOG_OP__
+ """
} finally {
sql "DROP DATABASE IF EXISTS test_binlog_changes_syntax_db"
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
