From f2633cebbc3cf1852227fde283b37072b8f68af3 Mon Sep 17 00:00:00 2001 From: Stefan Miklosovic Date: Wed, 21 Jan 2026 21:15:29 +0100 Subject: [PATCH] Do not serialize dropped columns to SSTable header after compaction patch by Stefan Miklosovic; reviewed by TBD for CASSANDRA-21000 --- .../cassandra/db/SerializationHeader.java | 20 ++- .../db/compaction/CursorCompactor.java | 10 ++ .../AbstractSerializationHeaderCQLTest.java | 151 ++++++++++++++++++ ...lizationHeaderCursorCompactionCQLTest.java | 33 ++++ ...lizationHeaderLegacyCompactionCQLTest.java | 32 ++++ 5 files changed, 245 insertions(+), 1 deletion(-) create mode 100644 test/unit/org/apache/cassandra/db/AbstractSerializationHeaderCQLTest.java create mode 100644 test/unit/org/apache/cassandra/db/SerializationHeaderCursorCompactionCQLTest.java create mode 100644 test/unit/org/apache/cassandra/db/SerializationHeaderLegacyCompactionCQLTest.java diff --git a/src/java/org/apache/cassandra/db/SerializationHeader.java b/src/java/org/apache/cassandra/db/SerializationHeader.java index 17b2888a89..bcb93c96c4 100644 --- a/src/java/org/apache/cassandra/db/SerializationHeader.java +++ b/src/java/org/apache/cassandra/db/SerializationHeader.java @@ -111,14 +111,32 @@ public class SerializationHeader // but rather on their stats stored in StatsMetadata that are fully accurate. EncodingStats.Collector stats = new EncodingStats.Collector(); RegularAndStaticColumns.Builder columns = RegularAndStaticColumns.builder(); + RegularAndStaticColumns currentSchemaColumns = metadata.regularAndStaticColumns(); // We need to order the SSTables by descending generation to be sure that we use latest column metadata. for (SSTableReader sstable : orderByDescendingGeneration(sstables)) { stats.updateTimestamp(sstable.getMinTimestamp()); stats.updateLocalDeletionTime(sstable.getMinLocalDeletionTime()); stats.updateTTL(sstable.getMinTTL()); - columns.addAll(sstable.header.columns()); + for (ColumnMetadata column : sstable.header.columns()) + { + // Include column in a header only if it is in the current schema. + // It is not enough to check if a column is not among dropped ones, + // because that would not cover the scenario when a column is dropped + // and re-added back before flush. Consider: + // + // insert into ks.tb (id, val) values (1, 2); + // alter table ks.tb drop val; + // alter table ks.tb add val int; + // + // If we backed our logic only by a check against a dropped column, + // then we would see that it was dropped (in metadata.droppedColumn) + // but we would omit the fact that it was actually re-added. + if (currentSchemaColumns.contains(column)) + columns.add(column); + } } + return new SerializationHeader(true, metadata, columns.build(), stats.get()); } diff --git a/src/java/org/apache/cassandra/db/compaction/CursorCompactor.java b/src/java/org/apache/cassandra/db/compaction/CursorCompactor.java index 3b4819c156..9f9a286971 100644 --- a/src/java/org/apache/cassandra/db/compaction/CursorCompactor.java +++ b/src/java/org/apache/cassandra/db/compaction/CursorCompactor.java @@ -636,6 +636,16 @@ public class CursorCompactor extends CompactionInfo.Holder ReusableLivenessInfo cellLiveness = cellCursor.cellLiveness; DataOutputBuffer tempCellBuffer = null; + if (!metadata().regularAndStaticColumns().contains(cellCursor.cellColumn)) + { + for (int i = 0; i < cellMergeLimit; i++) + { + if (sstableCursors[i].state() == CELL_VALUE_START) + sstableCursors[i].skipCellValue(); + } + return isRowDropped; + } + if (cellCursor.cellColumn.isComplex()) throw new UnsupportedOperationException("TODO: Not ready for complex cells."); if (cellCursor.cellColumn.isCounterColumn()) diff --git a/test/unit/org/apache/cassandra/db/AbstractSerializationHeaderCQLTest.java b/test/unit/org/apache/cassandra/db/AbstractSerializationHeaderCQLTest.java new file mode 100644 index 0000000000..8fd827f8d5 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/AbstractSerializationHeaderCQLTest.java @@ -0,0 +1,151 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.db; + +import org.junit.Ignore; +import org.junit.Test; + +import org.apache.cassandra.cql3.CQLTester; +import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.schema.ColumnMetadata; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +@Ignore +public abstract class AbstractSerializationHeaderCQLTest extends CQLTester +{ + @Test + public void testDroppedColumnPresenceInSerialisationHeader() + { + createTable("CREATE TABLE %s (id int PRIMARY KEY, a text, b text, c text)"); + execute("INSERT INTO %s (id, a, b, c) VALUES (1, 'a1', 'b1', 'c1')"); + flush(keyspace()); + alterTable("ALTER TABLE %s DROP b"); + + ColumnFamilyStore cfs = getCurrentColumnFamilyStore(); + assertEquals(1, cfs.getLiveSSTables().size()); + + SSTableReader sstable = cfs.getLiveSSTables().iterator().next(); + assertTrue("Column 'b' should be in the flushed SSTable header", + sstableHeaderContainsColumn(sstable, "b")); + + // same but alter before flush + + createTable("CREATE TABLE %s (id int PRIMARY KEY, a text, b text, c text)"); + execute("INSERT INTO %s (id, a, b, c) VALUES (1, 'a1', 'b1', 'c1')"); + alterTable("ALTER TABLE %s DROP b"); + flush(keyspace()); + + ColumnFamilyStore cfs2 = getCurrentColumnFamilyStore(); + assertEquals(1, cfs2.getLiveSSTables().size()); + + SSTableReader sstable2 = cfs2.getLiveSSTables().iterator().next(); + assertTrue("Column 'b' should be in the flushed SSTable header", + sstableHeaderContainsColumn(sstable2, "b")); + + // flush but without dropped column + + createTable("CREATE TABLE %s (id int PRIMARY KEY, a text, b text, c text)"); + // we are not populating column we go to drop on purpose + execute("INSERT INTO %s (id, a, c) VALUES (1, 'a1', 'c1')"); + alterTable("ALTER TABLE %s DROP b"); + flush(keyspace()); + + ColumnFamilyStore cfs3 = getCurrentColumnFamilyStore(); + assertEquals(1, cfs3.getLiveSSTables().size()); + + SSTableReader sstable3 = cfs3.getLiveSSTables().iterator().next(); + assertFalse("Column 'b' should not be in the flushed SSTable header", + sstableHeaderContainsColumn(sstable3, "b")); + } + + @Test + public void testDroppedColumnNotInCompactedSerialisationHeader() + { + createTable("CREATE TABLE %s (id int PRIMARY KEY, a text, b text, c text)"); + execute("INSERT INTO %s (id, a, b, c) VALUES (1, 'a1', 'b1', 'c1')"); + alterTable("ALTER TABLE %s DROP b"); + flush(keyspace()); + + ColumnFamilyStore cfs = getCurrentColumnFamilyStore(); + assertEquals(1, cfs.getLiveSSTables().size()); + assertTrue("Column 'b' should be in the flushed SSTable header", + sstableHeaderContainsColumn(cfs.getLiveSSTables().iterator().next(), "b")); + + compact(keyspace(), currentTable()); + + assertEquals(1, cfs.getLiveSSTables().size()); + + // column dropped for a header after compaction + assertFalse("Dropped column 'b' should not be in the compacted SSTable header", + sstableHeaderContainsColumn(cfs.getLiveSSTables().iterator().next(), "b")); + + assertTrue("Column 'a' should be in the compacted SSTable header", + sstableHeaderContainsColumn(cfs.getLiveSSTables().iterator().next(), "a")); + assertTrue("Column 'c' should be in the compacted SSTable header", + sstableHeaderContainsColumn(cfs.getLiveSSTables().iterator().next(), "c")); + + assertRows(execute("SELECT id, a, c FROM %s"), row(1, "a1", "c1")); + } + + @Test + public void testReAddedColumnInCompactedSerializationHeader() + { + createTable("CREATE TABLE %s (id int PRIMARY KEY, a text, b text, c text)"); + execute("INSERT INTO %s (id, a, b, c) VALUES (1, 'a1', 'b1', 'c1')"); + alterTable("ALTER TABLE %s DROP b"); + alterTable("ALTER TABLE %s ADD b text"); + flush(keyspace()); + + ColumnFamilyStore cfs = getCurrentColumnFamilyStore(); + assertEquals(1, cfs.getLiveSSTables().size()); + assertTrue("Column 'b' should be in the flushed SSTable header", + sstableHeaderContainsColumn(cfs.getLiveSSTables().iterator().next(), "b")); + + compact(keyspace(), currentTable()); + + assertEquals(1, cfs.getLiveSSTables().size()); + + // re-added column b + assertTrue("Dropped column 'b' should be in the compacted SSTable header", + sstableHeaderContainsColumn(cfs.getLiveSSTables().iterator().next(), "b")); + + assertTrue("Column 'a' should be in the compacted SSTable header", + sstableHeaderContainsColumn(cfs.getLiveSSTables().iterator().next(), "a")); + assertTrue("Column 'c' should be in the compacted SSTable header", + sstableHeaderContainsColumn(cfs.getLiveSSTables().iterator().next(), "c")); + + assertRows(execute("SELECT id, a, b, c FROM %s"), row(1, "a1", null, "c1")); + } + + /** + * Checks if a column with the given name exists in the SSTable header. + */ + private boolean sstableHeaderContainsColumn(SSTableReader sstable, String columnName) + { + for (ColumnMetadata column : sstable.header.columns()) + { + if (column.name.toString().equals(columnName)) + return true; + } + return false; + } +} diff --git a/test/unit/org/apache/cassandra/db/SerializationHeaderCursorCompactionCQLTest.java b/test/unit/org/apache/cassandra/db/SerializationHeaderCursorCompactionCQLTest.java new file mode 100644 index 0000000000..c75c595aac --- /dev/null +++ b/test/unit/org/apache/cassandra/db/SerializationHeaderCursorCompactionCQLTest.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.db; + +import org.junit.BeforeClass; + +import org.apache.cassandra.cql3.CQLTester; + +public class SerializationHeaderCursorCompactionCQLTest extends AbstractSerializationHeaderCQLTest +{ + @BeforeClass + public static void setUpClass() + { + System.setProperty("cassandra.cursor_compaction_enabled", "true"); + CQLTester.setUpClass(); + } +} diff --git a/test/unit/org/apache/cassandra/db/SerializationHeaderLegacyCompactionCQLTest.java b/test/unit/org/apache/cassandra/db/SerializationHeaderLegacyCompactionCQLTest.java new file mode 100644 index 0000000000..5b780f1c3d --- /dev/null +++ b/test/unit/org/apache/cassandra/db/SerializationHeaderLegacyCompactionCQLTest.java @@ -0,0 +1,32 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.db; + +import org.junit.BeforeClass; + +import org.apache.cassandra.cql3.CQLTester; + +public class SerializationHeaderLegacyCompactionCQLTest extends AbstractSerializationHeaderCQLTest +{ + @BeforeClass + public static void setUpClass() + { + System.setProperty("cassandra.cursor_compaction_enabled", "false"); + CQLTester.setUpClass(); + } +} \ No newline at end of file