Snapshot aware inserts

benchmark/micro/window/row_number.benchmark

@@ -0,0 +1,9 @@
+# name: benchmark/micro/window/row_number.benchmark
+# description: Check if the time to generate row_numbers after rewriting is way worse than a select * without rewriting. The reason is that we traverse the row groups to get the offset before actually handling the row groups and that can be slow
+# group: [window]
+
+load
+CREATE OR REPLACE TABLE foo AS SELECT range a FROM range(1300000000);
+
+run
+SELECT a, row_number() over() from foo;

src/catalog/catalog_entry/duck_table_entry.cpp

@@ -39,6 +39,19 @@ IndexStorageInfo GetIndexInfo(const IndexConstraintType type, const bool v1_0_0_
 	return index_info;
 }
 
+virtual_column_map_t DuckTableEntry::GetVirtualColumns() const {
+	virtual_column_map_t virtual_columns;
+	virtual_columns.insert(make_pair(COLUMN_IDENTIFIER_ROW_ID, TableColumn("rowid", LogicalType::ROW_TYPE)));
+	virtual_columns.insert(make_pair(COLUMN_IDENTIFIER_ROW_NUMBER, TableColumn("row_number", LogicalType::ROW_TYPE)));
+	return virtual_columns;
+}
+
+vector<column_t> DuckTableEntry::GetRowNumberColumns() const {
+	vector<column_t> result;
+	result.push_back(COLUMN_IDENTIFIER_ROW_NUMBER);
+	return result;
+}
+
 DuckTableEntry::DuckTableEntry(Catalog &catalog, SchemaCatalogEntry &schema, BoundCreateTableInfo &info,
                                shared_ptr<DataTable> inherited_storage)
     : TableCatalogEntry(catalog, schema, info.Base()), storage(std::move(inherited_storage)),
@@ -127,6 +140,9 @@ unique_ptr<BaseStatistics> DuckTableEntry::GetStatistics(ClientContext &context,
 	if (column_id == COLUMN_IDENTIFIER_ROW_ID) {
 		return nullptr;
 	}
+	if (column_id == COLUMN_IDENTIFIER_ROW_NUMBER) {
+		return nullptr;
+	}
 	auto &column = columns.GetColumn(LogicalIndex(column_id));
 	if (column.Generated()) {
 		return nullptr;

src/common/constants.cpp

@@ -11,6 +11,7 @@ const row_t MAX_ROW_ID = 36028797018960000ULL;       // 2^55
 const row_t MAX_ROW_ID_LOCAL = 72057594037920000ULL; // 2^56
 const column_t COLUMN_IDENTIFIER_ROW_ID = UINT64_C(18446744073709551615);
 const column_t COLUMN_IDENTIFIER_EMPTY = UINT64_C(18446744073709551614);
+const column_t COLUMN_IDENTIFIER_ROW_NUMBER = UINT64_C(18446744073709551613);
 const column_t VIRTUAL_COLUMN_START = UINT64_C(9223372036854775808); // 2^63
 const double PI = 3.141592653589793;
 
@@ -58,6 +59,10 @@ bool IsRowIdColumnId(column_t column_id) {
 	return column_id == COLUMN_IDENTIFIER_ROW_ID;
 }
 
+bool IsRowNumberColumnId(column_t column_id) {
+	return column_id == COLUMN_IDENTIFIER_ROW_NUMBER;
+}
+
 bool IsVirtualColumn(column_t column_id) {
 	return column_id >= VIRTUAL_COLUMN_START;
 }

src/common/enum_util.cpp

@@ -168,7 +168,6 @@
 #include "duckdb/planner/bound_result_modifier.hpp"
 #include "duckdb/planner/table_filter.hpp"
 #include "duckdb/storage/buffer/block_handle.hpp"
-#include "duckdb/storage/caching_file_system_wrapper.hpp"
 #include "duckdb/storage/compression/bitpacking.hpp"
 #include "duckdb/storage/magic_bytes.hpp"
 #include "duckdb/storage/statistics/base_statistics.hpp"
@@ -866,24 +865,6 @@ CTEMaterialize EnumUtil::FromString<CTEMaterialize>(const char *value) {
 	return static_cast<CTEMaterialize>(StringUtil::StringToEnum(GetCTEMaterializeValues(), 3, "CTEMaterialize", value));
 }
 
-const StringUtil::EnumStringLiteral *GetCachingModeValues() {
-	static constexpr StringUtil::EnumStringLiteral values[] {
-		{ static_cast<uint32_t>(CachingMode::ALWAYS_CACHE), "ALWAYS_CACHE" },
-		{ static_cast<uint32_t>(CachingMode::CACHE_REMOTE_ONLY), "CACHE_REMOTE_ONLY" }
-	};
-	return values;
-}
-
-template<>
-const char* EnumUtil::ToChars<CachingMode>(CachingMode value) {
-	return StringUtil::EnumToString(GetCachingModeValues(), 2, "CachingMode", static_cast<uint32_t>(value));
-}
-
-template<>
-CachingMode EnumUtil::FromString<CachingMode>(const char *value) {
-	return static_cast<CachingMode>(StringUtil::StringToEnum(GetCachingModeValues(), 2, "CachingMode", value));
-}
-
 const StringUtil::EnumStringLiteral *GetCatalogLookupBehaviorValues() {
 	static constexpr StringUtil::EnumStringLiteral values[] {
 		{ static_cast<uint32_t>(CatalogLookupBehavior::STANDARD), "STANDARD" },
@@ -947,22 +928,19 @@ const StringUtil::EnumStringLiteral *GetCheckpointAbortValues() {
 		{ static_cast<uint32_t>(CheckpointAbort::NO_ABORT), "NONE" },
 		{ static_cast<uint32_t>(CheckpointAbort::DEBUG_ABORT_BEFORE_TRUNCATE), "BEFORE_TRUNCATE" },
 		{ static_cast<uint32_t>(CheckpointAbort::DEBUG_ABORT_BEFORE_HEADER), "BEFORE_HEADER" },
-		{ static_cast<uint32_t>(CheckpointAbort::DEBUG_ABORT_AFTER_FREE_LIST_WRITE), "AFTER_FREE_LIST_WRITE" },
-		{ static_cast<uint32_t>(CheckpointAbort::DEBUG_ABORT_BEFORE_WAL_FINISH), "BEFORE_WAL_FINISH" },
-		{ static_cast<uint32_t>(CheckpointAbort::DEBUG_ABORT_BEFORE_MOVING_RECOVERY), "BEFORE_MOVING_RECOVERY" },
-		{ static_cast<uint32_t>(CheckpointAbort::DEBUG_ABORT_BEFORE_DELETING_CHECKPOINT_WAL), "BEFORE_DELETING_CHECKPOINT_WAL" }
+		{ static_cast<uint32_t>(CheckpointAbort::DEBUG_ABORT_AFTER_FREE_LIST_WRITE), "AFTER_FREE_LIST_WRITE" }
 	};
 	return values;
 }
 
 template<>
 const char* EnumUtil::ToChars<CheckpointAbort>(CheckpointAbort value) {
-	return StringUtil::EnumToString(GetCheckpointAbortValues(), 7, "CheckpointAbort", static_cast<uint32_t>(value));
+	return StringUtil::EnumToString(GetCheckpointAbortValues(), 4, "CheckpointAbort", static_cast<uint32_t>(value));
 }
 
 template<>
 CheckpointAbort EnumUtil::FromString<CheckpointAbort>(const char *value) {
-	return static_cast<CheckpointAbort>(StringUtil::StringToEnum(GetCheckpointAbortValues(), 7, "CheckpointAbort", value));
+	return static_cast<CheckpointAbort>(StringUtil::StringToEnum(GetCheckpointAbortValues(), 4, "CheckpointAbort", value));
 }
 
 const StringUtil::EnumStringLiteral *GetChunkInfoTypeValues() {
@@ -2968,6 +2946,7 @@ const StringUtil::EnumStringLiteral *GetMetricTypeValues() {
 		{ static_cast<uint32_t>(MetricType::OPTIMIZER_CTE_INLINING), "OPTIMIZER_CTE_INLINING" },
 		{ static_cast<uint32_t>(MetricType::OPTIMIZER_COMMON_SUBPLAN), "OPTIMIZER_COMMON_SUBPLAN" },
 		{ static_cast<uint32_t>(MetricType::OPTIMIZER_JOIN_ELIMINATION), "OPTIMIZER_JOIN_ELIMINATION" },
+		{ static_cast<uint32_t>(MetricType::OPTIMIZER_WINDOW_REWRITER), "OPTIMIZER_WINDOW_REWRITER" },
 		{ static_cast<uint32_t>(MetricType::ALL_OPTIMIZERS), "ALL_OPTIMIZERS" },
 		{ static_cast<uint32_t>(MetricType::CUMULATIVE_OPTIMIZER_TIMING), "CUMULATIVE_OPTIMIZER_TIMING" },
 		{ static_cast<uint32_t>(MetricType::PHYSICAL_PLANNER), "PHYSICAL_PLANNER" },
@@ -2980,15 +2959,6 @@ const StringUtil::EnumStringLiteral *GetMetricTypeValues() {
 	return values;
 }
 
-template<>
-const char* EnumUtil::ToChars<MetricType>(MetricType value) {
-	return StringUtil::EnumToString(GetMetricTypeValues(), 66, "MetricType", static_cast<uint32_t>(value));
-}
-
-template<>
-MetricType EnumUtil::FromString<MetricType>(const char *value) {
-	return static_cast<MetricType>(StringUtil::StringToEnum(GetMetricTypeValues(), 66, "MetricType", value));
-}
 
 const StringUtil::EnumStringLiteral *GetMultiFileColumnMappingModeValues() {
 	static constexpr StringUtil::EnumStringLiteral values[] {
@@ -2998,11 +2968,21 @@ const StringUtil::EnumStringLiteral *GetMultiFileColumnMappingModeValues() {
 	return values;
 }
 
+template<>
+MetricType EnumUtil::FromString<MetricType>(const char *value) {
+	return static_cast<MetricType>(StringUtil::StringToEnum(GetMetricTypeValues(), 67, "MetricType", value));
+}
+
 template<>
 const char* EnumUtil::ToChars<MultiFileColumnMappingMode>(MultiFileColumnMappingMode value) {
 	return StringUtil::EnumToString(GetMultiFileColumnMappingModeValues(), 2, "MultiFileColumnMappingMode", static_cast<uint32_t>(value));
 }
 
+template<>
+const char* EnumUtil::ToChars<MetricType>(MetricType value) {
+	return StringUtil::EnumToString(GetMetricTypeValues(), 67, "MetricType", static_cast<uint32_t>(value));
+}
+
 template<>
 MultiFileColumnMappingMode EnumUtil::FromString<MultiFileColumnMappingMode>(const char *value) {
 	return static_cast<MultiFileColumnMappingMode>(StringUtil::StringToEnum(GetMultiFileColumnMappingModeValues(), 2, "MultiFileColumnMappingMode", value));

src/common/enums/optimizer_type.cpp

@@ -45,6 +45,7 @@ static const DefaultOptimizerType internal_optimizer_types[] = {
     {"cte_inlining", OptimizerType::CTE_INLINING},
     {"common_subplan", OptimizerType::COMMON_SUBPLAN},
     {"join_elimination", OptimizerType::JOIN_ELIMINATION},
+    {"window_rewriter", OptimizerType::WINDOW_REWRITER},
     {nullptr, OptimizerType::INVALID}};
 
 string OptimizerTypeToString(OptimizerType type) {

src/function/table/read_duckdb.cpp

@@ -363,7 +363,7 @@ double DuckDBReader::GetProgressInFile(ClientContext &context) {
 }
 
 void DuckDBReader::AddVirtualColumn(column_t virtual_column_id) {
-	if (virtual_column_id != COLUMN_IDENTIFIER_ROW_ID) {
+	if (virtual_column_id != COLUMN_IDENTIFIER_ROW_ID && virtual_column_id != COLUMN_IDENTIFIER_ROW_NUMBER) {
 		throw InternalException("Unsupported virtual column id %d for duckdb reader", virtual_column_id);
 	}
 }
@@ -498,6 +498,7 @@ FileGlobInput DuckDBMultiFileInfo::GetGlobInput() {
 
 void DuckDBMultiFileInfo::GetVirtualColumns(ClientContext &, MultiFileBindData &, virtual_column_map_t &result) {
 	result.insert(make_pair(COLUMN_IDENTIFIER_ROW_ID, TableColumn("rowid", LogicalType::BIGINT)));
+	result.insert(make_pair(COLUMN_IDENTIFIER_ROW_NUMBER, TableColumn("row_number", LogicalType::BIGINT)));
 }
 
 void ReadDuckDBAddNamedParameters(TableFunction &table_function) {
@@ -511,6 +512,7 @@ static vector<column_t> DuckDBGetRowIdColumns(ClientContext &, optional_ptr<Func
 	vector<column_t> result;
 	result.emplace_back(MultiFileReader::COLUMN_IDENTIFIER_FILE_INDEX);
 	result.emplace_back(COLUMN_IDENTIFIER_ROW_ID);
+	result.emplace_back(COLUMN_IDENTIFIER_ROW_NUMBER);
 	return result;
 }
 

src/function/table/table_scan.cpp

@@ -24,7 +24,6 @@
 #include "duckdb/main/client_data.hpp"
 #include "duckdb/common/algorithm.hpp"
 #include "duckdb/planner/filter/optional_filter.hpp"
-#include "duckdb/planner/filter/selectivity_optional_filter.hpp"
 #include "duckdb/planner/filter/in_filter.hpp"
 #include "duckdb/planner/expression/bound_constant_expression.hpp"
 #include "duckdb/planner/expression/bound_comparison_expression.hpp"
@@ -41,6 +40,7 @@ struct TableScanLocalState : public LocalTableFunctionState {
 	//! The DataChunk containing all read columns.
 	//! This includes filter columns, which are immediately removed.
 	DataChunk all_columns;
+	idx_t row_number_count;
 };
 
 struct IndexScanLocalState : public LocalTableFunctionState {
@@ -71,6 +71,10 @@ static StorageIndex GetStorageIndex(TableCatalogEntry &table, const ColumnIndex
 		return StorageIndex();
 	}
 
+	if (column_id.IsRowNumberColumn()) {
+		return StorageIndex();
+	}
+
 	// The index of the base ColumnIndex is equal to the physical column index in the table
 	// for any child indices because the indices are already the physical indices.
 	// Only the top-level can have generated columns.
@@ -95,6 +99,10 @@ class TableScanGlobalState : public GlobalTableFunctionState {
 	vector<idx_t> projection_ids;
 	//! The types of all scanned columns.
 	vector<LogicalType> scanned_types;
+	//! The position of the row_number column
+	optional_idx row_number_col_index;
+	//! Synchronize changes to the global scan state.
+	mutex global_state_mutex;
 
 public:
 	virtual unique_ptr<LocalTableFunctionState> InitLocalState(ExecutionContext &context,
@@ -216,12 +224,6 @@ class DuckIndexScanState : public TableScanGlobalState {
 					storage.Fetch(tx, output, column_ids, local_vector, scan_count, l_state.fetch_state);
 				}
 				if (output.size() == 0) {
-					if (data_p.results_execution_mode == AsyncResultsExecutionMode::TASK_EXECUTOR) {
-						// We can avoid looping, and just return as appropriate
-						data_p.async_result = AsyncResultType::HAVE_MORE_OUTPUT;
-						return;
-					}
-
 					// output is empty, loop back, since there might be results to be picked up from LOCAL_STORAGE phase
 					continue;
 				}
@@ -286,6 +288,9 @@ class DuckTableScanState : public TableScanGlobalState {
 
 		vector<StorageIndex> storage_ids;
 		for (auto &col : input.column_indexes) {
+			if (col.IsRowNumberColumn()) {
+				continue; // initializte emit_row_numbers
+			}
 			storage_ids.push_back(GetStorageIndex(bind_data.table, col));
 		}
 
@@ -296,6 +301,12 @@ class DuckTableScanState : public TableScanGlobalState {
 
 		l_state->scan_state.Initialize(std::move(storage_ids), context.client, input.filters, input.sample_options);
 
+		if (state.scan_state.emit_row_numbers) {
+			auto &transaction = DuckTransaction::Get(context.client, bind_data.table.catalog);
+			l_state->scan_state.local_state.transaction = transaction;
+			l_state->scan_state.table_state.transaction = transaction;
+		}
+
 		storage.NextParallelScan(context.client, state, l_state->scan_state);
 		if (input.CanRemoveFilterColumns()) {
 			l_state->all_columns.Initialize(context.client, scanned_types);
@@ -309,7 +320,11 @@ class DuckTableScanState : public TableScanGlobalState {
 		auto &l_state = data_p.local_state->Cast<TableScanLocalState>();
 		l_state.scan_state.options.force_fetch_row = ClientConfig::GetConfig(context).force_fetch_row;
 
+		bool use_local = state.local_state.has_emitted_row_numbers;
 		do {
+			if (context.interrupted) {
+				throw InterruptException();
+			}
 			if (bind_data.is_create_index) {
 				storage.CreateIndexScan(l_state.scan_state, output,
 				                        TableScanType::TABLE_SCAN_COMMITTED_ROWS_OMIT_PERMANENTLY_DELETED);
@@ -321,27 +336,42 @@ class DuckTableScanState : public TableScanGlobalState {
 				storage.Scan(tx, output, l_state.scan_state);
 			}
 			if (output.size() > 0) {
-				return;
-			}
+				if (row_number_col_index.IsValid()) {
+					auto &row_number_vec = output.data[row_number_col_index.GetIndex()];
+					row_number_vec.SetVectorType(VectorType::FLAT_VECTOR);
+					auto row_number_data = FlatVector::GetData<row_t>(row_number_vec);
+					auto count = output.size();
+
+					if (l_state.scan_state.local_state.base_row_number == 0) {
+						auto break_here = 0;
+					}
+					idx_t base;
+					if (use_local && l_state.scan_state.local_state.base_row_number == 0) {
+						auto break_here = 0;
+					}
+					if (use_local && l_state.scan_state.local_state.base_row_number >= l_state.scan_state.table_state.base_row_number) {
+						base = l_state.scan_state.local_state.base_row_number + l_state.row_number_count;
+					} else {
+						base = l_state.scan_state.table_state.base_row_number + l_state.row_number_count;
+					}
+					if (l_state.scan_state.table_state.base_row_number + l_state.row_number_count > 500000) {
+						auto break_here = 0;
+					}
 
-			auto next = storage.NextParallelScan(context, state, l_state.scan_state);
-			if (data_p.results_execution_mode == AsyncResultsExecutionMode::TASK_EXECUTOR) {
-				// We can avoid looping, and just return as appropriate
-				if (!next) {
-					data_p.async_result = AsyncResultType::FINISHED;
-				} else {
-					data_p.async_result = AsyncResultType::HAVE_MORE_OUTPUT;
+					for (idx_t i = 0; i < count; i++) {
+						row_number_data[i] = static_cast<row_t>(base + i + 1);
+					}
+					l_state.row_number_count += count;
 				}
 				return;
 			}
+
+			auto next = storage.NextParallelScan(context, state, l_state.scan_state);
 			if (!next) {
 				return;
 			}
-
-			// Before looping back, check if we are interrupted
-			if (context.interrupted) {
-				throw InterruptException();
-			}
+			// One thread is assigned more than one batches. When we are done with one batch, we reset the counter.
+			l_state.row_number_count = 0;
 		} while (true);
 	}
 
@@ -391,6 +421,17 @@ unique_ptr<GlobalTableFunctionState> DuckTableScanInitGlobal(ClientContext &cont
 	}
 
 	storage.InitializeParallelScan(context, g_state->state);
+	for (idx_t i = 0; i < input.column_ids.size(); i++) {
+		if (input.column_ids[i] == COLUMN_IDENTIFIER_ROW_NUMBER) {
+			g_state->row_number_col_index = i;
+			g_state->state.scan_state.emit_row_numbers = true;
+			g_state->state.local_state.emit_row_numbers = true;
+			// g_state->state.local_state.initial_base_row_number = g_state->state.scan_state.base_row_number;
+			// g_state->state.scan_state.initial_base_row_number = g_state->state.scan_state.base_row_number;
+
+			break;
+		}
+	}
 	if (!input.CanRemoveFilterColumns()) {
 		return std::move(g_state);
 	}
@@ -399,7 +440,7 @@ unique_ptr<GlobalTableFunctionState> DuckTableScanInitGlobal(ClientContext &cont
 	auto &duck_table = bind_data.table.Cast<DuckTableEntry>();
 	const auto &columns = duck_table.GetColumns();
 	for (const auto &col_idx : input.column_indexes) {
-		if (col_idx.IsRowIdColumn()) {
+		if (col_idx.IsRowIdColumn() || col_idx.IsRowNumberColumn()) {
 			g_state->scanned_types.emplace_back(LogicalType::ROW_TYPE);
 		} else {
 			g_state->scanned_types.push_back(columns.GetColumn(col_idx.ToLogical()).Type());
@@ -467,9 +508,6 @@ bool ExtractComparisonsAndInFilters(TableFilter &filter, vector<reference<Consta
 		in_filters.push_back(filter.Cast<InFilter>());
 		return true;
 	}
-	case TableFilterType::BLOOM_FILTER: {
-		return true; // We can't use it for finding cmp/in filters, but we can just ignore it
-	}
 	case TableFilterType::CONJUNCTION_AND: {
 		auto &conjunction_and = filter.Cast<ConjunctionAndFilter>();
 		for (idx_t i = 0; i < conjunction_and.child_filters.size(); i++) {

src/include/duckdb/catalog/catalog_entry/duck_table_entry.hpp

@@ -55,6 +55,10 @@ class DuckTableEntry : public TableCatalogEntry {
 		return true;
 	}
 
+	//! Returns the virtual columns for this table
+	virtual_column_map_t GetVirtualColumns() const override;
+	virtual vector<column_t> GetRowNumberColumns() const;
+
 private:
 	unique_ptr<CatalogEntry> RenameColumn(ClientContext &context, RenameColumnInfo &info);
 	unique_ptr<CatalogEntry> RenameField(ClientContext &context, RenameFieldInfo &info);

src/include/duckdb/common/column_index.hpp

@@ -61,6 +61,9 @@ struct ColumnIndex {
 	bool IsRowIdColumn() const {
 		return index == COLUMN_IDENTIFIER_ROW_ID;
 	}
+	bool IsRowNumberColumn() const {
+		return index == COLUMN_IDENTIFIER_ROW_NUMBER;
+	}
 	bool IsEmptyColumn() const {
 		return index == COLUMN_IDENTIFIER_EMPTY;
 	}