diff --git a/benchmark/micro/window/row_number.benchmark b/benchmark/micro/window/row_number.benchmark new file mode 100644 index 000000000000..24a731158dee --- /dev/null +++ b/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; diff --git a/src/catalog/catalog_entry/duck_table_entry.cpp b/src/catalog/catalog_entry/duck_table_entry.cpp index dfecd0f16efb..bd10bd314386 100644 --- a/src/catalog/catalog_entry/duck_table_entry.cpp +++ b/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 DuckTableEntry::GetRowNumberColumns() const { + vector result; + result.push_back(COLUMN_IDENTIFIER_ROW_NUMBER); + return result; +} + DuckTableEntry::DuckTableEntry(Catalog &catalog, SchemaCatalogEntry &schema, BoundCreateTableInfo &info, shared_ptr inherited_storage) : TableCatalogEntry(catalog, schema, info.Base()), storage(std::move(inherited_storage)), @@ -127,6 +140,9 @@ unique_ptr 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; diff --git a/src/common/constants.cpp b/src/common/constants.cpp index 2bddd619532f..7e00aa9322e7 100644 --- a/src/common/constants.cpp +++ b/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; } diff --git a/src/common/enum_util.cpp b/src/common/enum_util.cpp index 707ea8ac69b5..2fa1bde3d553 100644 --- a/src/common/enum_util.cpp +++ b/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(const char *value) { return static_cast(StringUtil::StringToEnum(GetCTEMaterializeValues(), 3, "CTEMaterialize", value)); } -const StringUtil::EnumStringLiteral *GetCachingModeValues() { - static constexpr StringUtil::EnumStringLiteral values[] { - { static_cast(CachingMode::ALWAYS_CACHE), "ALWAYS_CACHE" }, - { static_cast(CachingMode::CACHE_REMOTE_ONLY), "CACHE_REMOTE_ONLY" } - }; - return values; -} - -template<> -const char* EnumUtil::ToChars(CachingMode value) { - return StringUtil::EnumToString(GetCachingModeValues(), 2, "CachingMode", static_cast(value)); -} - -template<> -CachingMode EnumUtil::FromString(const char *value) { - return static_cast(StringUtil::StringToEnum(GetCachingModeValues(), 2, "CachingMode", value)); -} - const StringUtil::EnumStringLiteral *GetCatalogLookupBehaviorValues() { static constexpr StringUtil::EnumStringLiteral values[] { { static_cast(CatalogLookupBehavior::STANDARD), "STANDARD" }, @@ -947,22 +928,19 @@ const StringUtil::EnumStringLiteral *GetCheckpointAbortValues() { { static_cast(CheckpointAbort::NO_ABORT), "NONE" }, { static_cast(CheckpointAbort::DEBUG_ABORT_BEFORE_TRUNCATE), "BEFORE_TRUNCATE" }, { static_cast(CheckpointAbort::DEBUG_ABORT_BEFORE_HEADER), "BEFORE_HEADER" }, - { static_cast(CheckpointAbort::DEBUG_ABORT_AFTER_FREE_LIST_WRITE), "AFTER_FREE_LIST_WRITE" }, - { static_cast(CheckpointAbort::DEBUG_ABORT_BEFORE_WAL_FINISH), "BEFORE_WAL_FINISH" }, - { static_cast(CheckpointAbort::DEBUG_ABORT_BEFORE_MOVING_RECOVERY), "BEFORE_MOVING_RECOVERY" }, - { static_cast(CheckpointAbort::DEBUG_ABORT_BEFORE_DELETING_CHECKPOINT_WAL), "BEFORE_DELETING_CHECKPOINT_WAL" } + { static_cast(CheckpointAbort::DEBUG_ABORT_AFTER_FREE_LIST_WRITE), "AFTER_FREE_LIST_WRITE" } }; return values; } template<> const char* EnumUtil::ToChars(CheckpointAbort value) { - return StringUtil::EnumToString(GetCheckpointAbortValues(), 7, "CheckpointAbort", static_cast(value)); + return StringUtil::EnumToString(GetCheckpointAbortValues(), 4, "CheckpointAbort", static_cast(value)); } template<> CheckpointAbort EnumUtil::FromString(const char *value) { - return static_cast(StringUtil::StringToEnum(GetCheckpointAbortValues(), 7, "CheckpointAbort", value)); + return static_cast(StringUtil::StringToEnum(GetCheckpointAbortValues(), 4, "CheckpointAbort", value)); } const StringUtil::EnumStringLiteral *GetChunkInfoTypeValues() { @@ -2968,6 +2946,7 @@ const StringUtil::EnumStringLiteral *GetMetricTypeValues() { { static_cast(MetricType::OPTIMIZER_CTE_INLINING), "OPTIMIZER_CTE_INLINING" }, { static_cast(MetricType::OPTIMIZER_COMMON_SUBPLAN), "OPTIMIZER_COMMON_SUBPLAN" }, { static_cast(MetricType::OPTIMIZER_JOIN_ELIMINATION), "OPTIMIZER_JOIN_ELIMINATION" }, + { static_cast(MetricType::OPTIMIZER_WINDOW_REWRITER), "OPTIMIZER_WINDOW_REWRITER" }, { static_cast(MetricType::ALL_OPTIMIZERS), "ALL_OPTIMIZERS" }, { static_cast(MetricType::CUMULATIVE_OPTIMIZER_TIMING), "CUMULATIVE_OPTIMIZER_TIMING" }, { static_cast(MetricType::PHYSICAL_PLANNER), "PHYSICAL_PLANNER" }, @@ -2980,15 +2959,6 @@ const StringUtil::EnumStringLiteral *GetMetricTypeValues() { return values; } -template<> -const char* EnumUtil::ToChars(MetricType value) { - return StringUtil::EnumToString(GetMetricTypeValues(), 66, "MetricType", static_cast(value)); -} - -template<> -MetricType EnumUtil::FromString(const char *value) { - return static_cast(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(const char *value) { + return static_cast(StringUtil::StringToEnum(GetMetricTypeValues(), 67, "MetricType", value)); +} + template<> const char* EnumUtil::ToChars(MultiFileColumnMappingMode value) { return StringUtil::EnumToString(GetMultiFileColumnMappingModeValues(), 2, "MultiFileColumnMappingMode", static_cast(value)); } +template<> +const char* EnumUtil::ToChars(MetricType value) { + return StringUtil::EnumToString(GetMetricTypeValues(), 67, "MetricType", static_cast(value)); +} + template<> MultiFileColumnMappingMode EnumUtil::FromString(const char *value) { return static_cast(StringUtil::StringToEnum(GetMultiFileColumnMappingModeValues(), 2, "MultiFileColumnMappingMode", value)); diff --git a/src/common/enums/optimizer_type.cpp b/src/common/enums/optimizer_type.cpp index f62af9626579..10a9acbc0560 100644 --- a/src/common/enums/optimizer_type.cpp +++ b/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) { diff --git a/src/function/table/read_duckdb.cpp b/src/function/table/read_duckdb.cpp index 4f8cfac66977..6ee3b67f2ee0 100644 --- a/src/function/table/read_duckdb.cpp +++ b/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 DuckDBGetRowIdColumns(ClientContext &, optional_ptr 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; } diff --git a/src/function/table/table_scan.cpp b/src/function/table/table_scan.cpp index 01902e3a541b..01601fb47ae3 100644 --- a/src/function/table/table_scan.cpp +++ b/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 projection_ids; //! The types of all scanned columns. vector 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 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 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(); 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_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(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 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 DuckTableScanInitGlobal(ClientContext &cont auto &duck_table = bind_data.table.Cast(); 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()); 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(); for (idx_t i = 0; i < conjunction_and.child_filters.size(); i++) { diff --git a/src/include/duckdb/catalog/catalog_entry/duck_table_entry.hpp b/src/include/duckdb/catalog/catalog_entry/duck_table_entry.hpp index 7f206a43d792..1a36f11451c6 100644 --- a/src/include/duckdb/catalog/catalog_entry/duck_table_entry.hpp +++ b/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 GetRowNumberColumns() const; + private: unique_ptr RenameColumn(ClientContext &context, RenameColumnInfo &info); unique_ptr RenameField(ClientContext &context, RenameFieldInfo &info); diff --git a/src/include/duckdb/common/column_index.hpp b/src/include/duckdb/common/column_index.hpp index b7bc3ae971f3..f473aa46051e 100644 --- a/src/include/duckdb/common/column_index.hpp +++ b/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; } diff --git a/src/include/duckdb/common/constants.hpp b/src/include/duckdb/common/constants.hpp index e9f816a17619..e1101f5d5cde 100644 --- a/src/include/duckdb/common/constants.hpp +++ b/src/include/duckdb/common/constants.hpp @@ -40,6 +40,8 @@ DUCKDB_API bool IsInvalidCatalog(const string &str); //! Special value used to signify the ROW ID of a table DUCKDB_API extern const column_t COLUMN_IDENTIFIER_ROW_ID; +//! Special value used to signify the ROW_NUMBER of a table +DUCKDB_API extern const column_t COLUMN_IDENTIFIER_ROW_NUMBER; //! Special value used to signify an empty column (used for e.g. COUNT(*)) DUCKDB_API extern const column_t COLUMN_IDENTIFIER_EMPTY; DUCKDB_API extern const column_t VIRTUAL_COLUMN_START; diff --git a/src/include/duckdb/common/enums/metric_type.hpp b/src/include/duckdb/common/enums/metric_type.hpp index 82208c895a85..33c9ea90bbc0 100644 --- a/src/include/duckdb/common/enums/metric_type.hpp +++ b/src/include/duckdb/common/enums/metric_type.hpp @@ -93,6 +93,7 @@ enum class MetricType : uint8_t { OPTIMIZER_CTE_INLINING, OPTIMIZER_COMMON_SUBPLAN, OPTIMIZER_JOIN_ELIMINATION, + OPTIMIZER_WINDOW_REWRITER, // PhaseTiming metrics ALL_OPTIMIZERS, CUMULATIVE_OPTIMIZER_TIMING, @@ -128,7 +129,7 @@ class MetricsUtils { static constexpr uint8_t END_OPERATOR = static_cast(MetricType::OPERATOR_TYPE); static constexpr uint8_t START_OPTIMIZER = static_cast(MetricType::OPTIMIZER_EXPRESSION_REWRITER); - static constexpr uint8_t END_OPTIMIZER = static_cast(MetricType::OPTIMIZER_JOIN_ELIMINATION); + static constexpr uint8_t END_OPTIMIZER = static_cast(MetricType::OPTIMIZER_WINDOW_REWRITER); static constexpr uint8_t START_PHASE_TIMING = static_cast(MetricType::ALL_OPTIMIZERS); static constexpr uint8_t END_PHASE_TIMING = static_cast(MetricType::PLANNER_BINDING); diff --git a/src/include/duckdb/common/enums/optimizer_type.hpp b/src/include/duckdb/common/enums/optimizer_type.hpp index 7f6864aacf39..82914f82183d 100644 --- a/src/include/duckdb/common/enums/optimizer_type.hpp +++ b/src/include/duckdb/common/enums/optimizer_type.hpp @@ -46,7 +46,8 @@ enum class OptimizerType : uint32_t { LATE_MATERIALIZATION, CTE_INLINING, COMMON_SUBPLAN, - JOIN_ELIMINATION + JOIN_ELIMINATION, + WINDOW_REWRITER, }; string OptimizerTypeToString(OptimizerType type); diff --git a/src/include/duckdb/function/table/table_scan.hpp b/src/include/duckdb/function/table/table_scan.hpp index 22407fff5f5a..4435cc70e7c8 100644 --- a/src/include/duckdb/function/table/table_scan.hpp +++ b/src/include/duckdb/function/table/table_scan.hpp @@ -30,6 +30,7 @@ struct TableScanBindData : public TableFunctionData { bool is_create_index; //! In what order to scan the row groups unique_ptr order_options; + idx_t row_number_projection_index = DConstants::INVALID_INDEX; public: bool Equals(const FunctionData &other_p) const override { diff --git a/src/include/duckdb/optimizer/window_rewriter.hpp b/src/include/duckdb/optimizer/window_rewriter.hpp new file mode 100644 index 000000000000..9c7e361f5425 --- /dev/null +++ b/src/include/duckdb/optimizer/window_rewriter.hpp @@ -0,0 +1,29 @@ +//===----------------------------------------------------------------------===// +// DuckDB +// +// duckdb/optimizer/window_rewriter.hpp +// +// +//===----------------------------------------------------------------------===// + +#pragma once +#include "duckdb/common/unique_ptr.hpp" +#include "duckdb/planner/logical_operator.hpp" +#include "duckdb/optimizer/column_binding_replacer.hpp" + +namespace duckdb { + +class Optimizer; + +class WindowRewriter { +public: + unique_ptr Optimize(unique_ptr op); + unique_ptr RewritePlan(unique_ptr op, ColumnBindingReplacer &replacer); + +private: + bool CanOptimize(LogicalOperator &op); + unique_ptr RewriteGet(unique_ptr op, ColumnBindingReplacer &replacer); + ColumnBindingReplacer replacer; +}; + +} // namespace duckdb diff --git a/src/include/duckdb/planner/table_binding.hpp b/src/include/duckdb/planner/table_binding.hpp index 836f52c417df..700408d6b664 100644 --- a/src/include/duckdb/planner/table_binding.hpp +++ b/src/include/duckdb/planner/table_binding.hpp @@ -57,6 +57,10 @@ struct Binding { static BindingAlias GetAlias(const string &explicit_alias, const StandardEntry &entry); static BindingAlias GetAlias(const string &explicit_alias, optional_ptr entry); + const case_insensitive_map_t &GetNameMap() const { + return name_map; + } + public: template TARGET &Cast() { diff --git a/src/include/duckdb/storage/storage_index.hpp b/src/include/duckdb/storage/storage_index.hpp index 38ba99976261..cc37e02a9c7c 100644 --- a/src/include/duckdb/storage/storage_index.hpp +++ b/src/include/duckdb/storage/storage_index.hpp @@ -61,6 +61,9 @@ struct StorageIndex { bool IsRowIdColumn() const { return index == DConstants::INVALID_INDEX; } + bool IsRowNumberColumn() const { + return index == COLUMN_IDENTIFIER_ROW_NUMBER; + } private: idx_t index; diff --git a/src/include/duckdb/storage/table/chunk_info.hpp b/src/include/duckdb/storage/table/chunk_info.hpp index db959f4cd117..3797bd251464 100644 --- a/src/include/duckdb/storage/table/chunk_info.hpp +++ b/src/include/duckdb/storage/table/chunk_info.hpp @@ -48,6 +48,8 @@ class ChunkInfo { virtual bool Fetch(TransactionData transaction, row_t row) = 0; virtual void CommitAppend(transaction_t commit_id, idx_t start, idx_t end) = 0; virtual idx_t GetCommittedDeletedCount(idx_t max_count) const = 0; + virtual idx_t GetCommittedDeletedCount(transaction_t min_start_time, transaction_t transaction_id, + idx_t max_count) = 0; virtual bool Cleanup(transaction_t lowest_transaction) const; virtual string ToString(idx_t max_count) const = 0; @@ -92,6 +94,7 @@ class ChunkConstantInfo : public ChunkInfo { bool Fetch(TransactionData transaction, row_t row) override; void CommitAppend(transaction_t commit_id, idx_t start, idx_t end) override; idx_t GetCommittedDeletedCount(idx_t max_count) const override; + idx_t GetCommittedDeletedCount(transaction_t min_start_id, transaction_t min_transaction_id, idx_t max_count) override; bool Cleanup(transaction_t lowest_transaction) const override; string ToString(idx_t max_count) const override; @@ -104,6 +107,8 @@ class ChunkConstantInfo : public ChunkInfo { template idx_t TemplatedGetSelVector(transaction_t start_time, transaction_t transaction_id, SelectionVector &sel_vector, idx_t max_count) const; + template + idx_t TemplatedGetCommittedDeleteCount(transaction_t start_time, transaction_t transaction_id, idx_t max_count) const; }; class ChunkVectorInfo : public ChunkInfo { @@ -126,6 +131,8 @@ class ChunkVectorInfo : public ChunkInfo { bool Cleanup(transaction_t lowest_transaction) const override; idx_t GetCommittedDeletedCount(idx_t max_count) const override; string ToString(idx_t max_count) const override; + idx_t GetCommittedDeletedCount(transaction_t min_start_time, transaction_t transaction_id, + idx_t max_count) override; void Append(idx_t start, idx_t end, transaction_t commit_id); @@ -149,6 +156,8 @@ class ChunkVectorInfo : public ChunkInfo { template idx_t TemplatedGetSelVector(transaction_t start_time, transaction_t transaction_id, SelectionVector &sel_vector, idx_t max_count) const; + template + idx_t TemplatedGetCommittedDeleteCount(transaction_t start_time, transaction_t transaction_id, idx_t max_count) const; IndexPointer GetInsertedPointer() const; IndexPointer GetDeletedPointer() const; diff --git a/src/include/duckdb/storage/table/row_group.hpp b/src/include/duckdb/storage/table/row_group.hpp index 37f89f847efe..0af5b2a7ec9f 100644 --- a/src/include/duckdb/storage/table/row_group.hpp +++ b/src/include/duckdb/storage/table/row_group.hpp @@ -174,6 +174,7 @@ class RowGroup : public SegmentBase { RowGroupWriteData WriteToDisk(RowGroupWriteInfo &info) const; //! Returns the number of committed rows (count - committed deletes) idx_t GetCommittedRowCount(); + idx_t GetCommittedRowCount(transaction_t min_start_time, transaction_t transaction_id); RowGroupWriteData WriteToDisk(RowGroupWriter &writer); RowGroupPointer Checkpoint(RowGroupWriteData write_data, RowGroupWriter &writer, TableStatistics &global_stats, idx_t row_group_start); diff --git a/src/include/duckdb/storage/table/row_group_collection.hpp b/src/include/duckdb/storage/table/row_group_collection.hpp index 32aa8af18f98..039c1a754d1e 100644 --- a/src/include/duckdb/storage/table/row_group_collection.hpp +++ b/src/include/duckdb/storage/table/row_group_collection.hpp @@ -8,6 +8,7 @@ #pragma once +#include "duckdb/storage/table/row_group_segment_tree.hpp" #include "duckdb/storage/table/row_group.hpp" #include "duckdb/storage/table/segment_tree.hpp" #include "duckdb/storage/statistics/column_statistics.hpp" diff --git a/src/include/duckdb/storage/table/row_version_manager.hpp b/src/include/duckdb/storage/table/row_version_manager.hpp index 8856ce57b2a7..5e65d26b1907 100644 --- a/src/include/duckdb/storage/table/row_version_manager.hpp +++ b/src/include/duckdb/storage/table/row_version_manager.hpp @@ -29,6 +29,7 @@ class RowVersionManager { return allocator; } idx_t GetCommittedDeletedCount(idx_t count); + idx_t GetCommittedDeletedCount(transaction_t min_start_time, transaction_t transaction_id, idx_t max_count); bool ShouldCheckpointRowGroup(transaction_t checkpoint_id, idx_t count); idx_t GetSelVector(TransactionData transaction, idx_t vector_idx, SelectionVector &sel_vector, idx_t max_count); diff --git a/src/include/duckdb/storage/table/scan_state.hpp b/src/include/duckdb/storage/table/scan_state.hpp index 3b17cfb68fbc..b754f83f221c 100644 --- a/src/include/duckdb/storage/table/scan_state.hpp +++ b/src/include/duckdb/storage/table/scan_state.hpp @@ -20,6 +20,7 @@ #include "duckdb/parser/parsed_data/sample_options.hpp" #include "duckdb/storage/storage_index.hpp" #include "duckdb/planner/table_filter_state.hpp" +#include "fmt/format.h" namespace duckdb { class AdaptiveFilter; @@ -213,6 +214,8 @@ class CollectionScanState { idx_t batch_index; //! The valid selection SelectionVector valid_sel; + idx_t base_row_number; + TransactionData transaction; RandomEngine random; @@ -266,6 +269,7 @@ class TableScanState { CollectionScanState table_state; //! Transaction-local scan state CollectionScanState local_state; + bool have_emitted_local_row_numbers = false; //! Options for scanning TableScanOptions options; //! Shared lock over the checkpoint to prevent checkpoints while reading @@ -304,6 +308,10 @@ struct ParallelCollectionScanState { idx_t vector_index; idx_t max_row; idx_t batch_index; + optional_idx base_row_number; + bool has_emitted_row_numbers = false; + optional_idx initial_base_row_number; + bool emit_row_numbers; atomic processed_rows; mutex lock; @@ -318,6 +326,7 @@ struct ParallelTableScanState { ParallelCollectionScanState local_state; //! Shared lock over the checkpoint to prevent checkpoints while reading shared_ptr checkpoint_lock; + optional_idx global_row_number; }; struct PrefetchState { diff --git a/src/optimizer/CMakeLists.txt b/src/optimizer/CMakeLists.txt index 0cb8de4129ed..b98a8f6d8d26 100644 --- a/src/optimizer/CMakeLists.txt +++ b/src/optimizer/CMakeLists.txt @@ -41,7 +41,8 @@ add_library_unity( topn_window_elimination.cpp unnest_rewriter.cpp sampling_pushdown.cpp - sum_rewriter.cpp) + sum_rewriter.cpp + window_rewriter.cpp) set(ALL_OBJECT_FILES ${ALL_OBJECT_FILES} $ PARENT_SCOPE) diff --git a/src/optimizer/optimizer.cpp b/src/optimizer/optimizer.cpp index 9bf4fcf8d592..87438f6fe63e 100644 --- a/src/optimizer/optimizer.cpp +++ b/src/optimizer/optimizer.cpp @@ -38,6 +38,7 @@ #include "duckdb/optimizer/unnest_rewriter.hpp" #include "duckdb/optimizer/late_materialization.hpp" #include "duckdb/optimizer/common_subplan_optimizer.hpp" +#include "duckdb/optimizer/window_rewriter.hpp" #include "duckdb/planner/binder.hpp" #include "duckdb/planner/planner.hpp" @@ -307,6 +308,12 @@ void Optimizer::RunBuiltInOptimizers() { JoinFilterPushdownOptimizer join_filter_pushdown(*this); join_filter_pushdown.VisitOperator(*plan); }); + + // Rewrite window functions to emit row_numbers in parallel + RunOptimizer(OptimizerType::WINDOW_REWRITER, [&]() { + WindowRewriter window_rewriter; + plan = window_rewriter.Optimize(std::move(plan)); + }); } unique_ptr Optimizer::Optimize(unique_ptr plan_p) { diff --git a/src/optimizer/window_rewriter.cpp b/src/optimizer/window_rewriter.cpp new file mode 100644 index 000000000000..dc69036d9ecd --- /dev/null +++ b/src/optimizer/window_rewriter.cpp @@ -0,0 +1,109 @@ +#include "duckdb/optimizer/window_rewriter.hpp" + +#include "duckdb/planner/operator/logical_get.hpp" +#include "duckdb/planner/expression/bound_window_expression.hpp" +#include "duckdb/optimizer/optimizer.hpp" +#include "duckdb/optimizer/column_binding_replacer.hpp" +#include "duckdb/planner/binder.hpp" +#include "duckdb/planner/operator/logical_projection.hpp" +#include "duckdb/planner/operator/logical_window.hpp" + +namespace duckdb { + +bool WindowRewriter::CanOptimize(LogicalOperator &op) { + // If the operator is a window function and its child is a get, check if optimization is possible + if (op.type != LogicalOperatorType::LOGICAL_WINDOW) { + return false; + } + + if (op.expressions.size() != 1) { + return false; + } + + auto &expression = op.expressions[0]; + auto &window_expr = expression->Cast(); + + // Try to optimize simple window functions, without partitions or ordering + if (!window_expr.partitions.empty() || !window_expr.orders.empty()) { + return false; + } + if (expression->type != ExpressionType::WINDOW_ROW_NUMBER) { + return false; + } + + // Should be followed by a get + auto &window_ch = op.children[0]; + if (window_ch->type != LogicalOperatorType::LOGICAL_GET) { + return false; + } + + // and can only be a seq_scan + auto &get = window_ch->Cast(); + if (get.virtual_columns.find(COLUMN_IDENTIFIER_ROW_NUMBER) == get.virtual_columns.end()) { + return false; + } + + return true; +} + +unique_ptr WindowRewriter::Optimize(unique_ptr op) { + LogicalOperator *root = op.get(); + op = RewritePlan(std::move(op), replacer); + + if (!replacer.replacement_bindings.empty()) { + replacer.VisitOperator(*root); + } + + return op; +} + +unique_ptr WindowRewriter::RewriteGet(unique_ptr op, + ColumnBindingReplacer &replacer) { + auto &window = op->Cast(); + auto &child = window.children[0]; + auto &get = child->Cast(); + + // Extend child LogicalGet output with virtual row_number column + auto column_ids = get.GetColumnIds(); + auto &types = get.types; + auto &projection_ids = get.projection_ids; + + column_ids.emplace_back(COLUMN_IDENTIFIER_ROW_NUMBER); + types.push_back(LogicalType::BIGINT); + + // In some queries, a dummy column may be projected (e.g., when a window function is used with a filter + // on a different column). This dummy column is later removed as it's not actually needed in the final output. + // In such cases, projections_ids is empty and we don't need to push a new projection id for the window function. + if (!projection_ids.empty()) { + projection_ids.push_back(column_ids.size() - 1); + } + get.SetColumnIds(std::move(column_ids)); + + const auto child_bindings = get.GetColumnBindings(); + + // Remove WINDOW and update bindings + const auto old_window_bindings = window.GetColumnBindings(); + + D_ASSERT(old_window_bindings.size() == child_bindings.size() + 1); + for (idx_t i = 0; i < child_bindings.size(); i++) { + replacer.replacement_bindings.emplace_back(old_window_bindings[i], child_bindings[i]); + } + replacer.replacement_bindings.emplace_back(old_window_bindings.back(), child_bindings.back()); + + replacer.stop_operator = child.get(); + return std::move(window.children[0]); +} + +unique_ptr WindowRewriter::RewritePlan(unique_ptr op, + ColumnBindingReplacer &replacer) { + if (CanOptimize(*op)) { + return RewriteGet(std::move(op), replacer); + } + + // Recurse into children + for (auto &child : op->children) { + child = RewritePlan(std::move(child), replacer); + } + return op; +} +} // namespace duckdb diff --git a/src/planner/table_binding.cpp b/src/planner/table_binding.cpp index c55d0be822b4..3820e9948cb7 100644 --- a/src/planner/table_binding.cpp +++ b/src/planner/table_binding.cpp @@ -161,6 +161,10 @@ TableBinding::TableBinding(const string &alias, vector types_p, vec // the empty column cannot be queried by the user continue; } + if (idx == COLUMN_IDENTIFIER_ROW_NUMBER) { + // the row_number column cannot be queried by the user + continue; + } if (name_map.find(name) == name_map.end()) { name_map[name] = idx; } diff --git a/src/storage/data_table.cpp b/src/storage/data_table.cpp index ba70e4188a23..073f955e8e45 100644 --- a/src/storage/data_table.cpp +++ b/src/storage/data_table.cpp @@ -281,7 +281,17 @@ bool DataTable::NextParallelScan(ClientContext &context, ParallelTableScanState if (row_groups->NextParallelScan(context, state.scan_state, scan_state.table_state)) { return true; } + auto &local_storage = LocalStorage::Get(context, db); + + if (!state.local_state.has_emitted_row_numbers) { + state.local_state.base_row_number = state.scan_state.base_row_number; + state.local_state.has_emitted_row_numbers = true; + scan_state.have_emitted_local_row_numbers = true; + } + if (state.local_state.has_emitted_row_numbers && !scan_state.have_emitted_local_row_numbers) { + auto break_here = 0; + } if (local_storage.NextParallelScan(context, *this, state.local_state, scan_state.local_state)) { return true; } else { diff --git a/src/storage/table/chunk_info.cpp b/src/storage/table/chunk_info.cpp index dfef0b4a1e2f..77471b71157c 100644 --- a/src/storage/table/chunk_info.cpp +++ b/src/storage/table/chunk_info.cpp @@ -7,6 +7,9 @@ #include "duckdb/common/serializer/memory_stream.hpp" #include "duckdb/transaction/delete_info.hpp" #include "duckdb/execution/index/fixed_size_allocator.hpp" +#include "fmt/format.h" + +#include namespace duckdb { @@ -73,6 +76,16 @@ idx_t ChunkConstantInfo::TemplatedGetSelVector(transaction_t start_time, transac return 0; } +template +idx_t ChunkConstantInfo::TemplatedGetCommittedDeleteCount(transaction_t start_time, transaction_t transaction_id, + idx_t max_count) const { + if (OP::UseInsertedVersion(start_time, transaction_id, insert_id) && + OP::UseDeletedVersion(start_time, transaction_id, delete_id)) { + return max_count; + } + return 0; +} + idx_t ChunkConstantInfo::GetSelVector(TransactionData transaction, SelectionVector &sel_vector, idx_t max_count) const { return TemplatedGetSelVector(transaction.start_time, transaction.transaction_id, sel_vector, max_count); @@ -83,6 +96,11 @@ idx_t ChunkConstantInfo::GetCommittedSelVector(transaction_t min_start_id, trans return TemplatedGetSelVector(min_start_id, min_transaction_id, sel_vector, max_count); } +idx_t ChunkConstantInfo::GetCommittedDeletedCount(transaction_t min_start_id, transaction_t min_transaction_id, + idx_t max_count) { + return TemplatedGetCommittedDeleteCount(min_start_id, min_transaction_id, max_count); +} + idx_t ChunkConstantInfo::GetCheckpointRowCount(TransactionData transaction, idx_t max_count) { if (TransactionVersionOperator::UseInsertedVersion(transaction.start_time, transaction.transaction_id, insert_id)) { return max_count; @@ -108,6 +126,17 @@ idx_t ChunkConstantInfo::GetCommittedDeletedCount(idx_t max_count) const { return delete_id < TRANSACTION_ID_START ? max_count : 0; } +// FIXME Find out when we use that and change accordingly +// idx_t ChunkConstantInfo::GetCommittedDeletedCount(transaction_t min_start_id, transaction_t min_transaction_id, idx_t +// max_count) { +// // return delete_id < TRANSACTION_ID_START ? max_count : 0; +// // A delete is visible to a snapshot if it was committed by another transaction after the snapshot started +// if ((delete_id < min_start_id || delete_id == min_transaction_id) ) { +// return max_count; +// } +// return 0; +// } + bool ChunkConstantInfo::Cleanup(transaction_t lowest_transaction) const { if (delete_id != NOT_DELETED_ID) { // the chunk info is labeled as deleted - we need to keep it around @@ -476,6 +505,127 @@ idx_t ChunkVectorInfo::GetCommittedDeletedCount(idx_t max_count) const { return delete_count; } +template +idx_t ChunkVectorInfo::TemplatedGetCommittedDeleteCount(transaction_t start_time, transaction_t transaction_id, + idx_t max_count) const { + if (HasConstantInsertionId()) { + if (!AnyDeleted()) { + // all tuples have the same inserted id: and no tuples were deleted + if (OP::UseInsertedVersion(start_time, transaction_id, ConstantInsertId())) { + return max_count; + } else { + return 0; + } + } + if (!OP::UseInsertedVersion(start_time, transaction_id, ConstantInsertId())) { + return 0; + } + // have to check deleted flag + // idx_t count = 0; + // auto segment = allocator.GetHandle(GetDeletedPointer()); + // auto deleted = segment.GetPtr(); + // for (idx_t i = 0; i < max_count; i++) { + // if (OP::UseDeletedVersion(start_time, transaction_id, deleted[i])) { + // // sel_vector.set_index(count++, i); + // count++; + // } + // } + // return count; + + auto segment = allocator.GetHandle(GetDeletedPointer()); + auto deleted = segment.GetPtr(); + + idx_t delete_count = 0; + for (idx_t i = 0; i < max_count; i++) { + // A delete is visible to a snapshot if it was committed by another transaction after the snapshot started + if ((deleted[i] < start_time || deleted[i] == transaction_id)) { + delete_count++; + } + } + return delete_count; + } + if (!AnyDeleted()) { + // have to check inserted flag + auto insert_segment = allocator.GetHandle(GetInsertedPointer()); + auto inserted = insert_segment.GetPtr(); + + idx_t count = 0; + for (idx_t i = 0; i < max_count; i++) { + if (OP::UseInsertedVersion(start_time, transaction_id, inserted[i])) { + // sel_vector.set_index(count++, i); + count++; + } + } + return count; + } + + idx_t count = 0; + // have to check both flags + auto insert_segment = allocator.GetHandle(GetInsertedPointer()); + auto inserted = insert_segment.GetPtr(); + + auto delete_segment = allocator.GetHandle(GetDeletedPointer()); + auto deleted = delete_segment.GetPtr(); + for (idx_t i = 0; i < max_count; i++) { + if (OP::UseInsertedVersion(start_time, transaction_id, inserted[i]) && + OP::UseDeletedVersion(start_time, transaction_id, deleted[i])) { + // sel_vector.set_index(count++, i); + count++; + } + } + return count; +} + +idx_t ChunkVectorInfo::GetCommittedDeletedCount(transaction_t start_time, transaction_t transaction_id, + idx_t max_count) { + if (!AnyDeleted()) { + // we wil come back to this, potentially inserted rows are deleted? Unsure + if (HasConstantInsertionId()) { + if (ConstantInsertId() < start_time || ConstantInsertId() == transaction_id) { + return max_count; + } + } + return 0; + } + + auto segment = allocator.GetHandle(GetDeletedPointer()); + auto deleted = segment.GetPtr(); + + idx_t delete_count = 0; + for (idx_t i = 0; i < max_count; i++) { + // A delete is visible to a snapshot if it was committed by another transaction after the snapshot started + if (!(deleted[i] < start_time || deleted[i] == transaction_id)) { + delete_count++; + } + } + return delete_count; + + // auto sel_vec = SelectionVector(); + // auto wat = TemplatedGetSelVector(start_time, transaction_id, sel_vec, max_count); + // auto break_here = 0; + // return wat; + // auto segment = allocator.GetHandle(GetDeletedPointer()); + // auto deleted = segment.GetPtr(); + // + // idx_t delete_count = 0; + // for (idx_t i = 0; i < max_count; i++) { + // // A delete is visible to a snapshot if it was committed by another transaction after the snapshot started + // // if ((deleted[i] < start_time || deleted[i] == transaction_id) ) { + // if (deleted[i] != transaction_id && deleted[i] < start_time) { + // delete_count++; + // } + // } + // return delete_count; + + // TODO: check this again? + // return TemplatedGetCommittedDeleteCount(start_time, transaction_id, max_count); + + // auto sel_vec = SelectionVector(); + // auto wat = TemplatedGetSelVector(start_time, transaction_id, sel_vec, max_count); + // auto break_here = 0; + // return wat; +} + void ChunkVectorInfo::Write(WriteStream &writer) const { SelectionVector sel(STANDARD_VECTOR_SIZE); transaction_t start_time = TRANSACTION_ID_START - 1; diff --git a/src/storage/table/row_group.cpp b/src/storage/table/row_group.cpp index f75903635cc9..4eeb3eb95c1b 100644 --- a/src/storage/table/row_group.cpp +++ b/src/storage/table/row_group.cpp @@ -265,7 +265,7 @@ void CollectionScanState::Initialize(const QueryContext &context, const vectorGetCommittedDeletedCount(count); } +idx_t RowGroup::GetCommittedRowCount(transaction_t start_time, transaction_t transaction_id) { + auto vinfo = GetVersionInfo(); + if (!vinfo) { + return count; + } + auto res = count - vinfo->GetCommittedDeletedCount(start_time, transaction_id, count); + return res; +} + bool RowGroup::HasUnloadedDeletes() const { if (deletes_pointers.empty()) { // no stored deletes at all diff --git a/src/storage/table/row_group_collection.cpp b/src/storage/table/row_group_collection.cpp index c6e0294aace4..512593b5dda9 100644 --- a/src/storage/table/row_group_collection.cpp +++ b/src/storage/table/row_group_collection.cpp @@ -264,6 +264,11 @@ bool RowGroupCollection::NextParallelScan(ClientContext &context, ParallelCollec auto row_start = state.current_row_group->GetRowStart(); collection = state.collection; row_group = state.current_row_group; + // if we are emitting row numbers, we need to set the current base row number + if (state.emit_row_numbers) { + scan_state.base_row_number = state.base_row_number.GetIndex(); + } + if (ClientConfig::GetConfig(context).verify_parallelism) { vector_index = state.vector_index; max_row = row_start + MinValue(current_row_group.count, @@ -274,11 +279,43 @@ bool RowGroupCollection::NextParallelScan(ClientContext &context, ParallelCollec state.current_row_group = state.GetNextRowGroup(*state.row_groups, *row_group).get(); state.vector_index = 0; } + // TODO: do this for verify_parallelism, OR do not do this optimization if verify_parallelism is set + // this could just throw an exception + if (state.emit_row_numbers) { + throw InvalidInputException("verify_parallelism is not supported with emitting row numbers"); + } } else { state.processed_rows += current_row_group.count; vector_index = 0; max_row = row_start + current_row_group.count; state.current_row_group = state.GetNextRowGroup(*state.row_groups, *row_group).get(); + // FIXME: this should not be GetCommittedRowCount but use the transaction id + if (state.emit_row_numbers) { + idx_t start = state.base_row_number.GetIndex(); + idx_t committed_row_count = current_row_group.GetCommittedRowCount( + scan_state.transaction.start_time, scan_state.transaction.transaction_id); + state.base_row_number = start + committed_row_count; + // Printer::PrintF("New base row number %d", state.base_row_number.GetIndex()); + // auto new_rn = state.base_row_number.GetIndex() + current_row_group.GetCommittedRowCount(); + // auto rn = state.base_row_number.GetIndex(); + // auto break_here = true; + + // for (idx_t r = 0, i = 0; r < state.collection->row_group_size ; r += STANDARD_VECTOR_SIZE, i++) { + // SelectionVector sel_vector; + // idx_t current_row = i * STANDARD_VECTOR_SIZE; + // auto max_count = MinValue(STANDARD_VECTOR_SIZE, state.collection->row_group_size - + // current_row); auto row_num = state.base_row_number.GetIndex() + + // current_row_group.GetSelVector(scan_state.transaction, i, sel_vector, max_count); + // state.base_row_number = row_num; + // } + + // SelectionVector sel_vector; + // idx_t current_row = state.batch_index * STANDARD_VECTOR_SIZE; + // auto max_count = MinValue(STANDARD_VECTOR_SIZE, state.collection->row_group_size - + // current_row); state.base_row_number = state.base_row_number.GetIndex() + + // current_row_group.GetSelVector(scan_state.transaction, scan_state.vector_index, sel_vector, + // max_count); + } } max_row = MinValue(max_row, state.max_row); scan_state.batch_index = ++state.batch_index; diff --git a/src/storage/table/row_version_manager.cpp b/src/storage/table/row_version_manager.cpp index 20d0ebed42b5..f6bb4f816bd3 100644 --- a/src/storage/table/row_version_manager.cpp +++ b/src/storage/table/row_version_manager.cpp @@ -29,6 +29,44 @@ idx_t RowVersionManager::GetCommittedDeletedCount(idx_t count) { return deleted_count; } +idx_t RowVersionManager::GetCommittedDeletedCount(transaction_t start_time, transaction_t transaction_id, idx_t count) { + lock_guard l(version_lock); + idx_t deleted_count = 0; + for (idx_t row_id = 0, row_batch = 0; row_id < count; row_id += STANDARD_VECTOR_SIZE, row_batch++) { + if (row_batch >= vector_info.size() || !vector_info[row_batch]) { + // deleted_count += STANDARD_VECTOR_SIZE; + continue; + } + idx_t max_count = MinValue(STANDARD_VECTOR_SIZE, count - row_id); + if (max_count == 0) { + break; + } + idx_t result = max_count; + if (vector_info[row_batch]) { + if (vector_info[row_batch]->type == ChunkInfoType::CONSTANT_INFO) { + auto &constant_vec_info = vector_info[row_batch]->Cast(); + SelectionVector sel; + TransactionData txn_data(transaction_id, start_time); + result = max_count - vector_info[row_batch]->GetSelVector(txn_data, sel, max_count); + } else { + result = + max_count - vector_info[row_batch]->GetCommittedDeletedCount(start_time, transaction_id, max_count); + } + } + deleted_count += result; + // ======= + // auto res = vector_info[i]->GetCommittedDeletedCount(start_time, transaction_id, max_count); + // deleted_count += res; + // >>>>>>> be30a58df64d5027e7de1cdbb40e81fe5df5ccfe + // SelectionVector sel_vector; + // TransactionData txn(transaction_id, start_time); + // + // deleted_count += vector_info[i]->GetSelVector(txn, sel_vector, max_count); + // auto break_here = true; + } + return deleted_count; +} + optional_ptr RowVersionManager::GetChunkInfo(idx_t vector_idx) { if (vector_idx >= vector_info.size()) { return nullptr; diff --git a/src/storage/table/scan_state.cpp b/src/storage/table/scan_state.cpp index 01314d7d18d0..b99efeca40db 100644 --- a/src/storage/table/scan_state.cpp +++ b/src/storage/table/scan_state.cpp @@ -176,7 +176,7 @@ TableScanOptions &CollectionScanState::GetOptions() { } ParallelCollectionScanState::ParallelCollectionScanState() - : collection(nullptr), current_row_group(nullptr), processed_rows(0) { + : collection(nullptr), current_row_group(nullptr), base_row_number(0), emit_row_numbers(false), processed_rows(0) { } optional_ptr> ParallelCollectionScanState::GetRootSegment(RowGroupSegmentTree &row_groups) const { @@ -196,7 +196,7 @@ ParallelCollectionScanState::GetNextRowGroup(RowGroupSegmentTree &row_groups, Se CollectionScanState::CollectionScanState(TableScanState &parent_p) : row_group(nullptr), vector_index(0), max_row_group_row(0), row_groups(nullptr), max_row(0), batch_index(0), - valid_sel(STANDARD_VECTOR_SIZE), random(-1), parent(parent_p) { + valid_sel(STANDARD_VECTOR_SIZE), transaction(0, 0), random(-1), parent(parent_p) { } optional_ptr> CollectionScanState::GetNextRowGroup(SegmentNode &row_group) const { diff --git a/test/optimizer/row_number_virtual_column.test b/test/optimizer/row_number_virtual_column.test new file mode 100644 index 000000000000..d998c941c4c2 --- /dev/null +++ b/test/optimizer/row_number_virtual_column.test @@ -0,0 +1,196 @@ +# name: test/optimizer/row_number_virtual_column.test +# description: Test row_number virtual column +# group: [optimizer] + +statement ok +CREATE OR REPLACE TABLE foo AS SELECT range a FROM range(130000); + +# Check that row_number is not accessible +statement error +SELECT a, row_number FROM foo; +---- +Binder Error: Referenced column "row_number" not found in FROM clause + +query I +SELECT a as row_number FROM foo limit 3; +---- +0 +1 +2 + +query I +select row_number() over() from foo where a < 3; +---- +1 +2 +3 + +statement ok +delete from foo where a < 5000; + +query III +select row_number() over(), a, rowid from foo order by a desc limit 1; +---- +125000 129999 129999 + +query III +select a, rowid, row_number() over() from foo order by a desc limit 1; +---- +129999 129999 125000 + +query II +explain select a,rowid, row_number() over() as x from foo order by a desc limit 1; +---- +physical_plan :.*WINDOW.* + +statement ok +select setseed(0.1); + +statement ok +create or replace table test as select (random()*10)::INT a, (random()*5)::INT b from range(10); + +statement ok +create or replace table test2 as select (random()*100)::INT c, (random()*5)::INT b from range(10); + +# Try with a window function and a Join + +query II +explain SELECT t.row_num, t.a, t2.c FROM ( SELECT a, b, ROW_NUMBER() OVER () AS row_num FROM test) t JOIN test2 t2 ON t.b = t2.b limit 10; +---- +physical_plan :.*WINDOW.* + + +statement ok +create table t1 as select range::FLOAT a, range b, range c, range d from range (3); + +query I +SELECT CASE WHEN row_number() OVER () = 2 THEN 2222::BIGINT ELSE 1111::BIGINT END FROM t1; +---- +1111 +2222 +1111 + +statement ok +CREATE OR REPLACE TABLE numbers AS SELECT range(1, 5) AS num; + +query II +SELECT row_number() OVER(), unnest(num) FROM numbers; +---- +1 1 +1 2 +1 3 +1 4 + +query II +SELECT row_number() OVER (), nums FROM (SELECT unnest(num) AS nums from numbers); +---- +1 1 +2 2 +3 3 +4 4 + +# Try with a join on the row_number + +statement ok +CREATE OR REPLACE TABLE foo AS SELECT range a from range(3); + +statement ok +CREATE OR REPLACE TABLE bar AS SELECT range b from range(2,5); + +query II +EXPLAIN SELECT f.row_num AS foo_row, b.row_num AS bar_row, f.a, b.b FROM (SELECT row_number() OVER () AS row_num, a FROM foo) f +JOIN (SELECT row_number() OVER () AS row_num, b FROM bar) b ON f.row_num = b.row_num; +---- +physical_plan :.*WINDOW.* + +statement ok +pragma disable_optimizer; + +statement ok +create table tbl as select range a from range(3); + +statement ok +pragma enable_optimizer; + +query II +EXPLAIN SELECT row_number + 1 FROM (SELECT row_number() OVER () AS row_number FROM tbl); +---- +physical_plan :.*WINDOW.* + + +# Try with multiple connections + +statement ok +SET immediate_transaction_mode=true; + +statement ok +CREATE TABLE big_tbl AS FROM range(1_000_000) t(i); + +statement ok con1 +BEGIN + +statement ok con2 +DELETE FROM big_tbl WHERE i%2=0; + +query III con1 +SELECT MIN(r), MAX(r), AVG(r) FROM ( + SELECT row_number() OVER () FROM big_tbl +) t(r) +---- +1 1000000 500000.5 + +query III con2 +SELECT MIN(r), MAX(r), AVG(r) FROM ( + SELECT row_number() OVER () FROM big_tbl +) t(r) +---- +1 500000 250000.5 + +statement ok con1 +COMMIT + + +# statement ok +# pragma disabled_optimizers="window_rewriter"; + +statement ok +pragma threads=2; + +statement ok +CREATE OR REPLACE TABLE big_tbl AS FROM range(1_000_000) t(i); + +statement ok con1 +BEGIN + +statement ok con1 +INSERT INTO big_tbl SELECT range FROM range(1_000_000); + +query III con1 +SELECT MIN(r), MAX(r), AVG(r) FROM ( + SELECT row_number() OVER () FROM big_tbl +) t(r) +---- +1 2000000 1000000.5 + +statement ok con1 +DELETE FROM big_tbl WHERE i%2=0; +# +# # statement ok con1 +# # DELETE FROM big_tbl WHERE i%2=0; + +query III con1 +SELECT MIN(r), MAX(r), AVG(r) FROM ( + SELECT row_number() OVER () FROM big_tbl +) t(r) +---- +1 1000000 500000.5 + +# statement ok con1 +# commit; +# +# query III +# SELECT MIN(r), MAX(r), AVG(r) FROM ( +# SELECT row_number() OVER () FROM big_tbl +# ) t(r) +# ---- +# 1 1500000 750000.5 \ No newline at end of file diff --git a/test/sql/pragma/profiling/test_all_profiling_settings.test b/test/sql/pragma/profiling/test_all_profiling_settings.test index 94966270a1b5..6b113e3a9441 100644 --- a/test/sql/pragma/profiling/test_all_profiling_settings.test +++ b/test/sql/pragma/profiling/test_all_profiling_settings.test @@ -78,6 +78,7 @@ SELECT unnest(res) FROM ( "OPTIMIZER_TOP_N_WINDOW_ELIMINATION": "true" "OPTIMIZER_UNNEST_REWRITER": "true" "OPTIMIZER_UNUSED_COLUMNS": "true" +"OPTIMIZER_WINDOW_REWRITER": "true" "PHYSICAL_PLANNER": "true" "PHYSICAL_PLANNER_COLUMN_BINDING": "true" "PHYSICAL_PLANNER_CREATE_PLAN": "true" diff --git a/test/sql/pragma/profiling/test_custom_profiling_using_groups.test b/test/sql/pragma/profiling/test_custom_profiling_using_groups.test index 156367048e84..e051f0874385 100644 --- a/test/sql/pragma/profiling/test_custom_profiling_using_groups.test +++ b/test/sql/pragma/profiling/test_custom_profiling_using_groups.test @@ -78,6 +78,7 @@ SELECT unnest(res) FROM ( "OPTIMIZER_TOP_N_WINDOW_ELIMINATION": "true" "OPTIMIZER_UNNEST_REWRITER": "true" "OPTIMIZER_UNUSED_COLUMNS": "true" +"OPTIMIZER_WINDOW_REWRITER": "true" "PHYSICAL_PLANNER": "true" "PHYSICAL_PLANNER_COLUMN_BINDING": "true" "PHYSICAL_PLANNER_CREATE_PLAN": "true" @@ -144,6 +145,7 @@ SELECT ALL_OPTIMIZERS, OPTIMIZER_TOP_N_WINDOW_ELIMINATION, OPTIMIZER_UNNEST_REWRITER, OPTIMIZER_UNUSED_COLUMNS, + OPTIMIZER_WINDOW_REWRITER, PHYSICAL_PLANNER, PHYSICAL_PLANNER_COLUMN_BINDING, PHYSICAL_PLANNER_CREATE_PLAN, @@ -464,6 +466,7 @@ SELECT unnest(res) FROM ( "OPTIMIZER_TOP_N_WINDOW_ELIMINATION": "true" "OPTIMIZER_UNNEST_REWRITER": "true" "OPTIMIZER_UNUSED_COLUMNS": "true" +"OPTIMIZER_WINDOW_REWRITER": "true" statement ok CREATE OR REPLACE TABLE metrics_output AS SELECT * FROM '__TEST_DIR__/profiling_output.json'; @@ -500,7 +503,8 @@ SELECT OPTIMIZER_BUILD_SIDE_PROBE_SIDE, OPTIMIZER_TOP_N, OPTIMIZER_TOP_N_WINDOW_ELIMINATION, OPTIMIZER_UNNEST_REWRITER, - OPTIMIZER_UNUSED_COLUMNS + OPTIMIZER_UNUSED_COLUMNS, + OPTIMIZER_WINDOW_REWRITER FROM metrics_output; statement ok @@ -559,6 +563,7 @@ SELECT unnest(res) FROM ( "OPTIMIZER_TOP_N_WINDOW_ELIMINATION": "true" "OPTIMIZER_UNNEST_REWRITER": "true" "OPTIMIZER_UNUSED_COLUMNS": "true" +"OPTIMIZER_WINDOW_REWRITER": "true" "PHYSICAL_PLANNER": "true" "PHYSICAL_PLANNER_COLUMN_BINDING": "true" "PHYSICAL_PLANNER_CREATE_PLAN": "true" @@ -604,6 +609,7 @@ SELECT ALL_OPTIMIZERS, OPTIMIZER_TOP_N_WINDOW_ELIMINATION, OPTIMIZER_UNNEST_REWRITER, OPTIMIZER_UNUSED_COLUMNS, + OPTIMIZER_WINDOW_REWRITER, PHYSICAL_PLANNER, PHYSICAL_PLANNER_COLUMN_BINDING, PHYSICAL_PLANNER_CREATE_PLAN, @@ -746,6 +752,7 @@ SELECT unnest(res) FROM ( "OPTIMIZER_TOP_N_WINDOW_ELIMINATION": "true" "OPTIMIZER_UNNEST_REWRITER": "true" "OPTIMIZER_UNUSED_COLUMNS": "true" +"OPTIMIZER_WINDOW_REWRITER": "true" "TOTAL_BYTES_READ": "true" "TOTAL_BYTES_WRITTEN": "true" "WAITING_TO_ATTACH_LATENCY": "true" @@ -792,6 +799,7 @@ SELECT ATTACH_LOAD_STORAGE_LATENCY, OPTIMIZER_TOP_N_WINDOW_ELIMINATION, OPTIMIZER_UNNEST_REWRITER, OPTIMIZER_UNUSED_COLUMNS, + OPTIMIZER_WINDOW_REWRITER, TOTAL_BYTES_READ, TOTAL_BYTES_WRITTEN, WAITING_TO_ATTACH_LATENCY, @@ -860,6 +868,7 @@ SELECT unnest(res) FROM ( "OPTIMIZER_TOP_N_WINDOW_ELIMINATION": "true" "OPTIMIZER_UNNEST_REWRITER": "true" "OPTIMIZER_UNUSED_COLUMNS": "true" +"OPTIMIZER_WINDOW_REWRITER": "true" "PHYSICAL_PLANNER": "true" "PHYSICAL_PLANNER_COLUMN_BINDING": "true" "PHYSICAL_PLANNER_CREATE_PLAN": "true" @@ -918,6 +927,7 @@ SELECT ALL_OPTIMIZERS, OPTIMIZER_TOP_N_WINDOW_ELIMINATION, OPTIMIZER_UNNEST_REWRITER, OPTIMIZER_UNUSED_COLUMNS, + OPTIMIZER_WINDOW_REWRITER, PHYSICAL_PLANNER, PHYSICAL_PLANNER_COLUMN_BINDING, PHYSICAL_PLANNER_CREATE_PLAN, diff --git a/test/sql/window/test_streaming_window.test b/test/sql/window/test_streaming_window.test index 5f2955b60c64..a818cd4ef3e2 100644 --- a/test/sql/window/test_streaming_window.test +++ b/test/sql/window/test_streaming_window.test @@ -54,7 +54,7 @@ physical_plan :.*STREAMING_WINDOW.* query TT explain select row_number() over (), i, j from integers ---- -physical_plan :.*STREAMING_WINDOW.* +physical_plan :.*STREAMING_WINDOW.* query TTT select row_number() over (), i, j from integers