natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Avoid re-initializing underlying iterator in LazilyInitializedUnfilteredRowIterator after closing 

Patch by marcuse; reviewed by Aleksey Yeschenko and Branimir Lambov for CASSANDRA-20972
This commit is contained in:
Marcus Eriksson 2025-10-08 15:44:23 +02:00
parent 98e7cd5d99
commit e721705152
8 changed files with 95 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
5.0.6
* Avoid re-initializing underlying iterator in LazilyInitializedUnfilteredRowIterator after closing (CASSANDRA-20972)
* Flush SAI segment builder when current SSTable writer is switched (CASSANDRA-20752)
* Throw RTE instead of FSError when RTE is thrown from FileUtis.write in TOCComponent (CASSANDRA-20917)
* Upgrade jackson-dataformat-yaml to 2.19.2 and snakeyaml to 2.1 (CASSANDRA-18875)

12
src/java/org/apache/cassandra/db/rows/LazilyInitializedUnfilteredRowIterator.java
View File

@ -32,8 +32,8 @@ import org.apache.cassandra.utils.AbstractIterator;
public abstract class LazilyInitializedUnfilteredRowIterator extends AbstractIterator<Unfiltered> implements UnfilteredRowIterator
{
private final DecoratedKey partitionKey;
private UnfilteredRowIterator iterator;
private boolean closed = false;
public LazilyInitializedUnfilteredRowIterator(DecoratedKey partitionKey)
{
@ -97,15 +97,17 @@ public abstract class LazilyInitializedUnfilteredRowIterator extends AbstractIte
public void close()
{
// don't use iterator == null as indicator if this is closed since some methods are called after the iterator is
// closed and maybeInit would re-initialize the underlying iterator in that case
closed = true;
if (iterator != null)
{
iterator.close();
iterator = null;
}
}
public boolean isOpen()
{
return iterator != null;
if (closed)
return false;
return iterator != null; // for backwards compatibility - if `maybeInit` has not been run on this class, consider it not open, for example SSTableExport seems to rely on this
}
}

47
test/distributed/org/apache/cassandra/distributed/test/DistinctReadTest.java Normal file
View File

@ -0,0 +1,47 @@
/*
* 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.distributed.test;
import java.io.IOException;
import org.junit.Test;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
public class DistinctReadTest extends TestBaseImpl
{
@Test
public void test() throws IOException
{
try (Cluster cluster = init(Cluster.build()
.withNodes(1)
.start()))
{
cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (id int, ck int, x int, PRIMARY KEY (id, ck))"));
cluster.coordinator(1).execute(withKeyspace("DELETE FROM %s.tbl USING TIMESTAMP 100 WHERE id = 1 AND ck < 10 "), ConsistencyLevel.ONE);
cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.tbl (id, ck, x) VALUES (1, 5, 7) USING TIMESTAMP 101"), ConsistencyLevel.ONE);
cluster.get(1).flush(KEYSPACE);
// all these failed before fix;
cluster.coordinator(1).execute(withKeyspace("select distinct id from %s.tbl where token(id) > " + Long.MIN_VALUE), ConsistencyLevel.ONE);
cluster.coordinator(1).execute(withKeyspace("select distinct id from %s.tbl where id > 0 allow filtering"), ConsistencyLevel.ONE);
cluster.coordinator(1).execute(withKeyspace("select id from %s.tbl where token(id) > " + Long.MIN_VALUE +" PER PARTITION LIMIT 1"), ConsistencyLevel.ONE);
}
}
}

38
test/unit/org/apache/cassandra/db/compaction/AntiCompactionTest.java
View File

@ -169,26 +169,28 @@ public class AntiCompactionTest
{
while (scanner.hasNext())
{
UnfilteredRowIterator row = scanner.next();
Token token = row.partitionKey().getToken();
if (sstable.isPendingRepair() && !sstable.isTransient())
{
assertTrue(fullContains.test(token));
assertFalse(transContains.test(token));
stats.pendingKeys++;
}
else if (sstable.isPendingRepair() && sstable.isTransient())
try (UnfilteredRowIterator row = scanner.next())
{
Token token = row.partitionKey().getToken();
if (sstable.isPendingRepair() && !sstable.isTransient())
{
assertTrue(fullContains.test(token));
assertFalse(transContains.test(token));
stats.pendingKeys++;
}
else if (sstable.isPendingRepair() && sstable.isTransient())
{
assertTrue(transContains.test(token));
assertFalse(fullContains.test(token));
stats.transKeys++;
}
else
{
assertFalse(fullContains.test(token));
assertFalse(transContains.test(token));
stats.unrepairedKeys++;
assertTrue(transContains.test(token));
assertFalse(fullContains.test(token));
stats.transKeys++;
}
else
{
assertFalse(fullContains.test(token));
assertFalse(transContains.test(token));
stats.unrepairedKeys++;
}
}
}
}

8
test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java
View File

@ -51,6 +51,7 @@ import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.ConfigurationException;
@ -286,7 +287,12 @@ public class LeveledCompactionStrategyTest
ISSTableScanner scanner = scanners.get(0);
// scan through to the end
while (scanner.hasNext())
scanner.next();
{
try (UnfilteredRowIterator ignored = scanner.next())
{
// just close the iterator
}
}
// scanner.getCurrentPosition should be equal to total bytes of L1 sstables
assertEquals(scanner.getCurrentPosition(), SSTableReader.getTotalUncompressedBytes(sstables));

6
test/unit/org/apache/cassandra/db/compaction/TTLExpiryTest.java
View File

@ -258,8 +258,10 @@ public class TTLExpiryTest
assertTrue(scanner.hasNext());
while(scanner.hasNext())
{
UnfilteredRowIterator iter = scanner.next();
assertEquals(Util.dk(noTTLKey), iter.partitionKey());
try (UnfilteredRowIterator iter = scanner.next())
{
assertEquals(Util.dk(noTTLKey), iter.partitionKey());
}
}
scanner.close();
}

6
test/unit/org/apache/cassandra/db/rows/ThrottledUnfilteredIteratorTest.java
View File

@ -214,8 +214,6 @@ public class ThrottledUnfilteredIteratorTest extends CQLTester
{
try (UnfilteredRowIterator rowIterator = scanner.next())
{
// only 1 partition data
assertFalse(scanner.hasNext());
List<Unfiltered> expectedUnfiltereds = new ArrayList<>();
rowIterator.forEachRemaining(expectedUnfiltereds::add);
@ -227,15 +225,17 @@ public class ThrottledUnfilteredIteratorTest extends CQLTester
assertTrue(scannerForThrottle.hasNext());
try (UnfilteredRowIterator rowIteratorForThrottle = scannerForThrottle.next())
{
assertFalse(scannerForThrottle.hasNext());
verifyThrottleIterator(expectedUnfiltereds,
rowIteratorForThrottle,
new ThrottledUnfilteredIterator(rowIteratorForThrottle, throttle),
throttle);
}
assertFalse(scannerForThrottle.hasNext());
}
}
}
// only 1 partition data
assertFalse(scanner.hasNext());
}
}

7
test/unit/org/apache/cassandra/io/sstable/SSTableScannerTest.java
View File

@ -345,7 +345,12 @@ public class SSTableScannerTest
// full range scan
ISSTableScanner scanner = sstable.getScanner();
for (int i = 4; i < 10; i++)
assertEquals(toKey(i), new String(scanner.next().partitionKey().getKey().array()));
{
try (UnfilteredRowIterator row = scanner.next())
{
assertEquals(toKey(i), new String(row.partitionKey().getKey().array()));
}
}
scanner.close();