From 5d6b8b3d19ca2c72ddf0abb89d91d286c271714d Mon Sep 17 00:00:00 2001 From: DuckLabs GitHub Bot Date: Tue, 18 Aug 2026 02:54:46 +0000 Subject: [PATCH] Update vendored DuckDB sources to 11dc00c898 --- CMakeLists.txt | 1 + .../extension/icu/datetime/calendar.cpp | 67 +- .../extension/icu/datetime/gregorian.cpp | 6 +- .../icu/datetime/include/calendar.hpp | 5 - .../extension/icu/datetime/include/coptic.hpp | 3 - .../icu/datetime/include/gregorian.hpp | 9 - .../icu/datetime/include/japanese.hpp | 3 - src/duckdb/extension/icu/icu-datesub.cpp | 32 + .../src/common/operator/cast_operators.cpp | 2 +- .../vector_operations/scalar_executor.cpp | 26 + .../scan/physical_column_data_scan.cpp | 13 +- .../execution/operator/set/physical_cte.cpp | 39 +- .../set/physical_recursive_cte_runtime.cpp | 2 +- .../execution/operator/set/physical_union.cpp | 4 +- .../function/table/version/pragma_version.cpp | 6 +- .../include/duckdb/common/smaller_binary.hpp | 24 + .../vector_operations/binary_executor.hpp | 710 +++------------ .../vector_operations/scalar_executor.hpp | 813 ++++++++++++++++++ .../vector_operations/unary_executor.hpp | 321 ++----- .../vector_operations/variadic_executor.hpp | 266 ++---- .../operator/join/physical_hash_join.hpp | 6 + .../execution/operator/set/physical_cte.hpp | 4 + .../duckdb/execution/physical_operator.hpp | 4 + .../include/duckdb/parallel/meta_pipeline.hpp | 22 +- .../src/include/duckdb/parallel/pipeline.hpp | 32 +- .../parallel/pipeline_broadcast_exchange.hpp | 4 + .../duckdb/parallel/pipeline_schedule.hpp | 25 + .../optimizer/rule/timestamp_comparison.cpp | 19 +- src/duckdb/src/parallel/executor.cpp | 13 +- src/duckdb/src/parallel/meta_pipeline.cpp | 41 +- src/duckdb/src/parallel/pipeline.cpp | 378 +++++++- .../parallel/pipeline_broadcast_exchange.cpp | 37 +- src/duckdb/src/parallel/pipeline_schedule.cpp | 126 ++- .../table/column_data_checkpointer.cpp | 41 +- 34 files changed, 1961 insertions(+), 1143 deletions(-) create mode 100644 src/duckdb/src/common/vector_operations/scalar_executor.cpp create mode 100644 src/duckdb/src/include/duckdb/common/vector_operations/scalar_executor.hpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 35b501123..8ece7097e 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -172,6 +172,7 @@ set(DUCKDB_SRC_FILES src/duckdb/src/common/vector_operations/is_distinct_from.cpp src/duckdb/src/common/vector_operations/null_operations.cpp src/duckdb/src/common/vector_operations/numeric_inplace_operators.cpp + src/duckdb/src/common/vector_operations/scalar_executor.cpp src/duckdb/src/common/vector_operations/vector_cast.cpp src/duckdb/src/common/vector_operations/vector_copy.cpp src/duckdb/src/common/vector_operations/vector_hash.cpp diff --git a/src/duckdb/extension/icu/datetime/calendar.cpp b/src/duckdb/extension/icu/datetime/calendar.cpp index 57b2dbd2e..0497ad583 100644 --- a/src/duckdb/extension/icu/datetime/calendar.cpp +++ b/src/duckdb/extension/icu/datetime/calendar.cpp @@ -832,6 +832,7 @@ void FieldCalendar::AddChecked(CalendarField field, int32_t amount) { switch (field) { case CAL_ERA: { const auto era = GetChecked(CAL_ERA); + const auto year = GetChecked(CAL_YEAR); if (failed) { return; } @@ -842,18 +843,35 @@ void FieldCalendar::AddChecked(CalendarField field, int32_t amount) { } Set(CAL_ERA, sum); PinField(CAL_ERA); + // keep the same year within the new era + Set(CAL_YEAR, year); + PinField(CAL_DATE); return; } - case CAL_YEAR: - case CAL_YEAR_WOY: - // in an era that counts backwards, later years have smaller numbers - if (GetChecked(CAL_ERA) == 0 && IsEra0CountingBackward()) { - if (!TryMultiply(amount, -1, amount)) { - Fail(); - return; - } + case CAL_YEAR: { + // extended years continue across era boundaries + const auto year = GetChecked(CAL_EXTENDED_YEAR); + int32_t sum; + if (failed || !TryAdd(year, amount, sum)) { + Fail(); + return; } - DUCKDB_EXPLICIT_FALLTHROUGH; + Set(CAL_EXTENDED_YEAR, sum); + PinField(CAL_DATE); + return; + } + case CAL_YEAR_WOY: { + // week-based year numbers continue across era boundaries + const auto year = GetChecked(CAL_YEAR_WOY); + int32_t sum; + if (failed || !TryAdd(year, amount, sum)) { + Fail(); + return; + } + Set(CAL_YEAR_WOY, sum); + PinField(CAL_DATE); + return; + } case CAL_EXTENDED_YEAR: case CAL_MONTH: case CAL_ORDINAL_MONTH: { @@ -1049,6 +1067,37 @@ int32_t FieldCalendar::FieldDifference(int64_t target, CalendarField field) { failed = false; int32_t min = 0; const auto start = GetTimeChecked(); + if (field == CAL_ERA) { + // eras can start partway through a year, so only count complete eras between the two dates + const auto start_era = GetChecked(CAL_ERA); + SetTimeChecked(target); + const auto target_era = GetChecked(CAL_ERA); + const auto difference = int64_t(target_era) - start_era; + if (failed) { + return 0; + } + if (difference < NumericLimits::Minimum() || difference > NumericLimits::Maximum()) { + Fail(); + return 0; + } + // start by applying the difference between the two era numbers + auto result = int32_t(difference); + SetTimeChecked(start); + AddChecked(field, result); + const auto reached = GetTimeChecked(); + if (failed) { + return 0; + } + // check whether applying the era number difference reaches the target without overshooting it + if (reached == target || (start < target ? reached < target : reached > target)) { + return result; + } + result += start < target ? -1 : 1; + // leave the calendar at the date reached by the returned difference + SetTimeChecked(start); + AddChecked(field, result); + return failed ? 0 : result; + } // most differences are close to what the average length of the field predicts, which saves // the search below from having to bracket the answer from scratch diff --git a/src/duckdb/extension/icu/datetime/gregorian.cpp b/src/duckdb/extension/icu/datetime/gregorian.cpp index 40eced819..8865b6d4f 100644 --- a/src/duckdb/extension/icu/datetime/gregorian.cpp +++ b/src/duckdb/extension/icu/datetime/gregorian.cpp @@ -177,11 +177,7 @@ int32_t GregorianCalendar::HandleGetExtendedYear() { return InternalGet(CAL_YEAR, EPOCH_YEAR); } case CAL_YEAR_WOY: { - auto year_woy = InternalGet(CAL_YEAR_WOY); - if (InternalGet(CAL_ERA, AD) == BC) { - year_woy = 1 - year_woy; - } - return HandleGetExtendedYearFromWeekFields(year_woy, InternalGet(CAL_WEEK_OF_YEAR)); + return HandleGetExtendedYearFromWeekFields(InternalGet(CAL_YEAR_WOY), InternalGet(CAL_WEEK_OF_YEAR)); } default: return EPOCH_YEAR; diff --git a/src/duckdb/extension/icu/datetime/include/calendar.hpp b/src/duckdb/extension/icu/datetime/include/calendar.hpp index 3ea35532c..2dcd2b1a9 100644 --- a/src/duckdb/extension/icu/datetime/include/calendar.hpp +++ b/src/duckdb/extension/icu/datetime/include/calendar.hpp @@ -125,11 +125,6 @@ class FieldCalendar : public Calendar { //! A table of groups, terminated by a null group using ResolutionTable = const ResolutionGroup *; - //! Whether the years of the first era count backwards, as in the Gregorian BC era - virtual bool IsEra0CountingBackward() const { - return false; - } - const TimeZone &GetTimeZone() const { return *zone; } diff --git a/src/duckdb/extension/icu/datetime/include/coptic.hpp b/src/duckdb/extension/icu/datetime/include/coptic.hpp index f6dc6caa9..74fa0f366 100644 --- a/src/duckdb/extension/icu/datetime/include/coptic.hpp +++ b/src/duckdb/extension/icu/datetime/include/coptic.hpp @@ -48,9 +48,6 @@ class CopticCalendar : public CopticEthiopicCalendar { const char *GetType() const override { return "coptic"; } - bool IsEra0CountingBackward() const override { - return true; - } unique_ptr Copy() const override { return unique_ptr(new CopticCalendar(*this)); } diff --git a/src/duckdb/extension/icu/datetime/include/gregorian.hpp b/src/duckdb/extension/icu/datetime/include/gregorian.hpp index 9f4c2d33c..7ae9cc16c 100644 --- a/src/duckdb/extension/icu/datetime/include/gregorian.hpp +++ b/src/duckdb/extension/icu/datetime/include/gregorian.hpp @@ -39,9 +39,6 @@ class GregorianCalendar : public FieldCalendar { const char *GetType() const override { return "gregorian"; } - bool IsEra0CountingBackward() const override { - return true; - } unique_ptr Copy() const override { return unique_ptr(new GregorianCalendar(*this)); } @@ -86,9 +83,6 @@ class BuddhistCalendar : public GregorianCalendar { const char *GetType() const override { return "buddhist"; } - bool IsEra0CountingBackward() const override { - return false; - } unique_ptr Copy() const override { return unique_ptr(new BuddhistCalendar(*this)); } @@ -141,9 +135,6 @@ class ISO8601Calendar : public GregorianCalendar { const char *GetType() const override { return "iso8601"; } - bool IsEra0CountingBackward() const override { - return false; - } unique_ptr Copy() const override { return unique_ptr(new ISO8601Calendar(*this)); } diff --git a/src/duckdb/extension/icu/datetime/include/japanese.hpp b/src/duckdb/extension/icu/datetime/include/japanese.hpp index af68bfe70..1a1ee8d86 100644 --- a/src/duckdb/extension/icu/datetime/include/japanese.hpp +++ b/src/duckdb/extension/icu/datetime/include/japanese.hpp @@ -35,9 +35,6 @@ class JapaneseCalendar : public GregorianCalendar { const char *GetType() const override { return "japanese"; } - bool IsEra0CountingBackward() const override { - return false; - } unique_ptr Copy() const override { return unique_ptr(new JapaneseCalendar(*this)); } diff --git a/src/duckdb/extension/icu/icu-datesub.cpp b/src/duckdb/extension/icu/icu-datesub.cpp index 329599235..3aa483c14 100644 --- a/src/duckdb/extension/icu/icu-datesub.cpp +++ b/src/duckdb/extension/icu/icu-datesub.cpp @@ -8,6 +8,18 @@ namespace duckdb { struct ICUCalendarSub : public ICUDateFunc { + static unique_ptr Bind(BindScalarFunctionInput &input) { + auto part_value = input.TryGetConstant(0); + if (part_value && !part_value->IsNull()) { + DatePartSpecifier part; + if (TryGetDatePartSpecifier(part_value->GetValue(), part) && part == DatePartSpecifier::ERA) { + // date_sub is not monotone for eras because era boundaries can occur partway through a year + input.GetBoundFunction().SetArgProperties({}); + } + } + return ICUDateFunc::Bind(input); + } + // ICU only has 32 bit precision for date parts, so it can overflow a high resolution. // Since there is no difference between ICU and the obvious calculations, // we make these using the DuckDB internal type. @@ -189,6 +201,20 @@ ICUDateFunc::part_sub_t ICUDateFunc::SubtractFactory(DatePartSpecifier type) { // MS-SQL differences can be computed using ICU by truncating both arguments // to the desired part precision and then applying ICU subtraction/difference struct ICUCalendarDiff : public ICUDateFunc { + static int64_t DifferenceEra(Calendar *calendar, timestamp_tz_t start_date, timestamp_tz_t end_date) { + SetTime(calendar, start_date); + const auto start_era = ExtractField(calendar, CAL_ERA); + SetTime(calendar, end_date); + return int64_t(ExtractField(calendar, CAL_ERA)) - start_era; + } + + static int64_t DifferenceEra(Calendar *calendar, timestamp_tz_ns_t start_date, timestamp_tz_ns_t end_date) { + SetTimeNS(calendar, start_date); + const auto start_era = ExtractField(calendar, CAL_ERA); + SetTimeNS(calendar, end_date); + return int64_t(ExtractField(calendar, CAL_ERA)) - start_era; + } + static timestamp_tz_t TruncateForDiff(Calendar *calendar, timestamp_tz_t date, part_trunc_t trunc_func) { auto micros = SetTime(calendar, date); trunc_func(calendar, micros); @@ -249,6 +275,9 @@ struct ICUCalendarDiff : public ICUDateFunc { BinaryExecutor::Execute( startdate_arg, enddate_arg, result, [&](T start_date, T end_date) -> optional { if (start_date.IsFinite() && end_date.IsFinite()) { + if (part == DatePartSpecifier::ERA) { + return DifferenceEra(calendar, start_date, end_date); + } return DifferenceFunc(calendar, start_date, end_date, trunc_func, sub_func); } else { return nullopt; @@ -261,6 +290,9 @@ struct ICUCalendarDiff : public ICUDateFunc { [&](string_t specifier, T start_date, T end_date) -> optional { if (start_date.IsFinite() && end_date.IsFinite()) { const auto part = GetDatePartSpecifier(specifier.GetString()); + if (part == DatePartSpecifier::ERA) { + return DifferenceEra(calendar, start_date, end_date); + } auto trunc_func = DiffTruncationFactory(part); auto sub_func = SubtractFactory(part); return DifferenceFunc(calendar, start_date, end_date, trunc_func, sub_func); diff --git a/src/duckdb/src/common/operator/cast_operators.cpp b/src/duckdb/src/common/operator/cast_operators.cpp index 8204208ee..98ac1c496 100644 --- a/src/duckdb/src/common/operator/cast_operators.cpp +++ b/src/duckdb/src/common/operator/cast_operators.cpp @@ -2826,7 +2826,7 @@ bool DoubleToDecimalCast(SRC input, DST &result, CastParameters ¶meters, uin return false; } // For some reason PG does not use statistical rounding here (even though it _does_ for integers...) - result = Cast::Operation(static_cast(roundedValue)); + result = Cast::Operation(roundedValue); return true; } diff --git a/src/duckdb/src/common/vector_operations/scalar_executor.cpp b/src/duckdb/src/common/vector_operations/scalar_executor.cpp new file mode 100644 index 000000000..929e9f206 --- /dev/null +++ b/src/duckdb/src/common/vector_operations/scalar_executor.cpp @@ -0,0 +1,26 @@ +#include "duckdb/common/vector_operations/scalar_executor.hpp" + +namespace duckdb { + +bool ScalarExecutor::PrepareGenericResultValidity(const UnifiedVectorFormat *formats, idx_t format_count, + Vector &result, idx_t count, bool preserve_result_validity, + bool adds_nulls) { + bool inputs_can_have_null = false; + for (idx_t input_idx = 0; input_idx < format_count; input_idx++) { + if (formats[input_idx].validity.CanHaveNull()) { + inputs_can_have_null = true; + break; + } + } + + auto &result_validity = FlatVector::ValidityMutable(result); + if (inputs_can_have_null || !preserve_result_validity) { + result_validity.Reset(count); + } else if (adds_nulls && result_validity.CanHaveNull()) { + ValidityMask preserved(result_validity, count); + result_validity.Initialize(preserved); + } + return inputs_can_have_null; +} + +} // namespace duckdb diff --git a/src/duckdb/src/execution/operator/scan/physical_column_data_scan.cpp b/src/duckdb/src/execution/operator/scan/physical_column_data_scan.cpp index 7694b300d..5608ca4b6 100644 --- a/src/duckdb/src/execution/operator/scan/physical_column_data_scan.cpp +++ b/src/duckdb/src/execution/operator/scan/physical_column_data_scan.cpp @@ -246,11 +246,16 @@ void PhysicalColumnDataScan::BuildPipelines(Pipeline ¤t, MetaPipeline &met auto &source = cte_source->Cast(); // Prefer direct fanout. Buffered exchange is only used when it can avoid full materialization or // when the consumer can stop early; otherwise this scan reads the materialized working table. + auto has_blocking_dependencies = !current.GetDependencies().empty(); + if (cte.CanRegisterDirectConsumer(current) && + (!cte.cte_body_has_side_effects || has_blocking_dependencies)) { + D_ASSERT(collection); + current.AddDependency(cte_dependency); + state.AddExternalInputCandidate(current, *this, *cte_source, cte_dependency, cte, source.consumer_idx); + state.SetPipelineSource(current, *this); + return; + } if (cte.TryRegisterDirectConsumer(current, source.consumer_idx)) { - auto current_pipeline = current.shared_from_this(); - current.SetExternalInput(); - current.AddExternalFinishDependency(cte_dependency); - cte_dependency->AddDataflowDependency(current_pipeline); DUCKDB_LOG(current.GetClientContext(), PhysicalOperatorLogType, cte, "PhysicalCTE", "SelectConsumer", {{"consumer", to_string(source.consumer_idx)}, {"mode", "DIRECT"}}); state.SetPipelineSource(current, *cte_source); diff --git a/src/duckdb/src/execution/operator/set/physical_cte.cpp b/src/duckdb/src/execution/operator/set/physical_cte.cpp index d8597d1f4..8173b1362 100644 --- a/src/duckdb/src/execution/operator/set/physical_cte.cpp +++ b/src/duckdb/src/execution/operator/set/physical_cte.cpp @@ -418,12 +418,16 @@ void PhysicalCTE::BuildPipelines(Pipeline ¤t, MetaPipeline &meta_pipeline) } children[1].get().BuildPipelines(current, meta_pipeline); - pipeline_selection_state = CTEPipelineSelectionState::RESOLVED; + auto has_unresolved_consumers = exchange && exchange->GetConsumerSummary().unresolved > 0; + pipeline_selection_state = + has_unresolved_consumers ? CTEPipelineSelectionState::UNRESOLVED : CTEPipelineSelectionState::RESOLVED; + bool dependency_added = false; if (exchange && !UseStreamingExchange()) { - // All exchange consumers were converted to materialized scans during pipeline construction. + // Keep the materialized fallback until graph-dependent consumer selection has completed. auto cte_pipeline = child_meta_pipeline.GetBasePipeline(); current.AddDependency(cte_pipeline); + dependency_added = true; } if (exchange && last_child_ptr && !current.HasDataflowDependencies()) { for (auto &side_effect_pipeline : side_effect_pipelines) { @@ -435,19 +439,40 @@ void PhysicalCTE::BuildPipelines(Pipeline ¤t, MetaPipeline &meta_pipeline) exchange ? DataflowDependencyMode::SKIP_CONFLICTING : DataflowDependencyMode::INCLUDE); } + if (has_unresolved_consumers) { + state.AddCTEPipelineSelection(*this, current, child_meta_pipeline.GetBasePipeline(), dependency_added); + } } bool PhysicalCTE::TryRegisterDirectConsumer(Pipeline &pipeline, idx_t consumer_idx) { - if (!exchange) { - return false; - } - if (!exchange->TryRegisterDirectConsumer(pipeline, consumer_idx)) { + if (!CanRegisterDirectConsumer(pipeline)) { return false; } - RegisterBatchPreference(pipeline); + pipeline.SetExternalInput(GetProducerPipelines()); + RegisterDirectConsumer(pipeline, consumer_idx); return true; } +bool PhysicalCTE::CanRegisterDirectConsumer(Pipeline &pipeline) const { + return exchange && exchange->CanRegisterDirectConsumer(pipeline); +} + +void PhysicalCTE::RegisterDirectConsumer(Pipeline &pipeline, idx_t consumer_idx) { + D_ASSERT(exchange); + exchange->SelectDirectConsumer(pipeline, consumer_idx); + RegisterBatchPreference(pipeline); +} + +vector> PhysicalCTE::GetProducerPipelines() const { + D_ASSERT(exchange); + return exchange->GetProducerPipelines(); +} + +void PhysicalCTE::SetPipelineSelectionResolved() { + D_ASSERT(pipeline_selection_state == CTEPipelineSelectionState::UNRESOLVED); + pipeline_selection_state = CTEPipelineSelectionState::RESOLVED; +} + bool PhysicalCTE::ShouldUseBufferedConsumer(Pipeline &pipeline) const { if (!exchange) { return false; diff --git a/src/duckdb/src/execution/operator/set/physical_recursive_cte_runtime.cpp b/src/duckdb/src/execution/operator/set/physical_recursive_cte_runtime.cpp index 8ab2bf51f..6deb7a37b 100644 --- a/src/duckdb/src/execution/operator/set/physical_recursive_cte_runtime.cpp +++ b/src/duckdb/src/execution/operator/set/physical_recursive_cte_runtime.cpp @@ -605,7 +605,7 @@ FindInvariantRecursiveMetaPipelines(const vector> &meta if (!depends_on_variant) { for (auto &entry : meta_pipeline->GetDependencies()) { for (auto &dependency : entry.second) { - auto dep_entry = pipeline_to_meta_pipeline.find(dependency.get()); + auto dep_entry = pipeline_to_meta_pipeline.find(dependency.pipeline.get()); if (dep_entry == pipeline_to_meta_pipeline.end()) { continue; } diff --git a/src/duckdb/src/execution/operator/set/physical_union.cpp b/src/duckdb/src/execution/operator/set/physical_union.cpp index 1002036f3..3b373b449 100644 --- a/src/duckdb/src/execution/operator/set/physical_union.cpp +++ b/src/duckdb/src/execution/operator/set/physical_union.cpp @@ -80,7 +80,9 @@ void PhysicalUnion::BuildPipelines(Pipeline ¤t, MetaPipeline &meta_pipelin if (order_matters || can_saturate_threads) { // we add dependencies if order matters: union_pipeline comes after all pipelines created by building // current - dependencies = meta_pipeline.AddDependenciesFrom(union_pipeline, union_pipeline, false); + auto dependency_type = + order_matters ? MetaPipelineDependencyType::REQUIRED : MetaPipelineDependencyType::OPTIONAL_DEPENDENCY; + dependencies = meta_pipeline.AddDependenciesFrom(union_pipeline, union_pipeline, false, dependency_type); // we also add dependencies if the LHS child can saturate all available threads // in that case, we recursively make all RHS children depend on the LHS. // This prevents breadth-first plan evaluation diff --git a/src/duckdb/src/function/table/version/pragma_version.cpp b/src/duckdb/src/function/table/version/pragma_version.cpp index 0750f4452..c12043f0d 100644 --- a/src/duckdb/src/function/table/version/pragma_version.cpp +++ b/src/duckdb/src/function/table/version/pragma_version.cpp @@ -1,5 +1,5 @@ #ifndef DUCKDB_PATCH_VERSION -#define DUCKDB_PATCH_VERSION "0-alpha38069" +#define DUCKDB_PATCH_VERSION "0-alpha38143" #endif #ifndef DUCKDB_MINOR_VERSION #define DUCKDB_MINOR_VERSION 0 @@ -8,10 +8,10 @@ #define DUCKDB_MAJOR_VERSION 2 #endif #ifndef DUCKDB_VERSION -#define DUCKDB_VERSION "v2.0.0-alpha38069" +#define DUCKDB_VERSION "v2.0.0-alpha38143" #endif #ifndef DUCKDB_SOURCE_ID -#define DUCKDB_SOURCE_ID "bcd78db0ec" +#define DUCKDB_SOURCE_ID "11dc00c898" #endif #include "duckdb/function/table/system_functions.hpp" #include "duckdb/main/database.hpp" diff --git a/src/duckdb/src/include/duckdb/common/smaller_binary.hpp b/src/duckdb/src/include/duckdb/common/smaller_binary.hpp index fa52f1190..c25b00266 100644 --- a/src/duckdb/src/include/duckdb/common/smaller_binary.hpp +++ b/src/duckdb/src/include/duckdb/common/smaller_binary.hpp @@ -52,6 +52,14 @@ #define DUCKDB_SB_FEATURE_unary_executor_flat DUCKDB_SB_DEFAULT // group: vector_specialization #endif +#ifndef DUCKDB_SB_FEATURE_unary_executor_select_flat +#define DUCKDB_SB_FEATURE_unary_executor_select_flat DUCKDB_SB_DEFAULT // group: vector_specialization +#endif + +#ifndef DUCKDB_SB_FEATURE_unary_executor_select_flags +#define DUCKDB_SB_FEATURE_unary_executor_select_flags DUCKDB_SB_DEFAULT // group: vector_specialization +#endif + #ifndef DUCKDB_SB_FEATURE_binary_executor_flat #define DUCKDB_SB_FEATURE_binary_executor_flat DUCKDB_SB_DEFAULT // group: vector_specialization #endif @@ -64,6 +72,22 @@ #define DUCKDB_SB_FEATURE_binary_executor_select_flags DUCKDB_SB_DEFAULT // group: vector_specialization #endif +#ifndef DUCKDB_SB_FEATURE_binary_executor_generic_nullable +#define DUCKDB_SB_FEATURE_binary_executor_generic_nullable DUCKDB_SB_DEFAULT // group: vector_specialization +#endif + +#ifndef DUCKDB_SB_FEATURE_variadic_executor_select_flat +#define DUCKDB_SB_FEATURE_variadic_executor_select_flat DUCKDB_SB_DEFAULT // group: vector_specialization +#endif + +#ifndef DUCKDB_SB_FEATURE_variadic_executor_flat +#define DUCKDB_SB_FEATURE_variadic_executor_flat DUCKDB_SB_DEFAULT // group: vector_specialization +#endif + +#ifndef DUCKDB_SB_FEATURE_variadic_executor_select_flags +#define DUCKDB_SB_FEATURE_variadic_executor_select_flags DUCKDB_SB_DEFAULT // group: vector_specialization +#endif + #ifndef DUCKDB_SB_FEATURE_aggregate_executor_flat #define DUCKDB_SB_FEATURE_aggregate_executor_flat DUCKDB_SB_DEFAULT // group: vector_specialization #endif diff --git a/src/duckdb/src/include/duckdb/common/vector_operations/binary_executor.hpp b/src/duckdb/src/include/duckdb/common/vector_operations/binary_executor.hpp index 8724b467b..1e3ed8971 100644 --- a/src/duckdb/src/include/duckdb/common/vector_operations/binary_executor.hpp +++ b/src/duckdb/src/include/duckdb/common/vector_operations/binary_executor.hpp @@ -3,43 +3,38 @@ // // duckdb/common/vector_operations/binary_executor.hpp // -// //===----------------------------------------------------------------------===// #pragma once -#include "duckdb/common/exception.hpp" #include "duckdb/common/operator/comparison_operators.hpp" #include "duckdb/common/optional.hpp" -#include "duckdb/common/types/vector.hpp" -#include "duckdb/common/vector/constant_vector.hpp" -#include "duckdb/common/vector/flat_vector.hpp" +#include "duckdb/common/smaller_binary.hpp" +#include "duckdb/common/vector_operations/scalar_executor.hpp" #include "duckdb/common/vector_operations/vector_operations.hpp" -#include "duckdb/common/smaller_binary.hpp" #include +#include namespace duckdb { -//! Complement-fold trait for comparison selection. On the NO_NULL path a comparison op's selection -//! is the exact complement of its complement op's selection (no NULL third bucket). -//! Note LessThanEquals is not handled since that's done upstream in the general case (see -//! VectorOperations::LessThan[Equals]) template struct ComparisonSelectComplement { static constexpr bool FOLD = false; }; + template <> struct ComparisonSelectComplement { static constexpr bool FOLD = true; using COMPLEMENT = Equals; static constexpr bool SWAP_OPERANDS = false; }; + template <> struct ComparisonSelectComplement { static constexpr bool FOLD = true; using COMPLEMENT = GreaterThan; - static constexpr bool SWAP_OPERANDS = true; // GreaterThanEquals(a,b) == !GreaterThan(b,a) + static constexpr bool SWAP_OPERANDS = true; }; struct DefaultNullCheckOperator { @@ -51,22 +46,22 @@ struct DefaultNullCheckOperator { struct BinaryStandardOperatorWrapper { template - static inline RESULT_TYPE Operation(FUNC fun, LEFT_TYPE left, RIGHT_TYPE right, ValidityMask &mask, idx_t idx) { + static inline RESULT_TYPE Operation(FUNC &fun, LEFT_TYPE left, RIGHT_TYPE right, ValidityMask &mask, idx_t idx) { return OP::template Operation(left, right); } - static bool AddsNulls() { + static constexpr bool AddsNulls() { return false; } }; struct BinarySingleArgumentOperatorWrapper { template - static inline RESULT_TYPE Operation(FUNC fun, LEFT_TYPE left, RIGHT_TYPE right, ValidityMask &mask, idx_t idx) { + static inline RESULT_TYPE Operation(FUNC &fun, LEFT_TYPE left, RIGHT_TYPE right, ValidityMask &mask, idx_t idx) { return OP::template Operation(left, right); } - static bool AddsNulls() { + static constexpr bool AddsNulls() { return false; } }; @@ -74,7 +69,7 @@ struct BinarySingleArgumentOperatorWrapper { template struct BinaryLambdaWrapper { template - static inline RESULT_TYPE Operation(FUNC fun, LEFT_TYPE left, RIGHT_TYPE right, ValidityMask &mask, idx_t idx) { + static inline RESULT_TYPE Operation(FUNC &fun, LEFT_TYPE left, RIGHT_TYPE right, ValidityMask &mask, idx_t idx) { if constexpr (ADDS_NULLS) { auto result = fun(left, right); if (!result.has_value()) { @@ -87,239 +82,137 @@ struct BinaryLambdaWrapper { } } - static bool AddsNulls() { + static constexpr bool AddsNulls() { return ADDS_NULLS; } }; -struct BinaryExecutor { -#if !DUCKDB_SMALLER_BINARY(binary_executor_flat) - template - static void ExecuteFlatLoop(const LEFT_TYPE *__restrict ldata, const RIGHT_TYPE *__restrict rdata, - RESULT_TYPE *__restrict result_data, idx_t count, ValidityMask &mask, FUNC fun) { - if (!LEFT_CONSTANT) { - ASSERT_RESTRICT(ldata, ldata + count, result_data, result_data + count); - } - if (!RIGHT_CONSTANT) { - ASSERT_RESTRICT(rdata, rdata + count, result_data, result_data + count); - } +template +struct BinaryScalarAdapter { + static constexpr bool ADDS_NULLS = CAN_ADD_NULLS; - if (mask.CanHaveNull()) { - idx_t base_idx = 0; - auto entry_count = ValidityMask::EntryCount(count); - for (idx_t entry_idx = 0; entry_idx < entry_count; entry_idx++) { - auto validity_entry = mask.GetValidityEntry(entry_idx); - idx_t next = MinValue(base_idx + ValidityMask::BITS_PER_VALUE, count); - if (ValidityMask::AllValid(validity_entry)) { - // all valid: perform operation - for (; base_idx < next; base_idx++) { - auto lentry = ldata[LEFT_CONSTANT ? 0 : base_idx]; - auto rentry = rdata[RIGHT_CONSTANT ? 0 : base_idx]; - result_data[base_idx] = - OPWRAPPER::template Operation( - fun, lentry, rentry, mask, base_idx); - } - } else if (ValidityMask::NoneValid(validity_entry)) { - // nothing valid: skip all - base_idx = next; - continue; - } else { - // partially valid: need to check individual elements for validity - idx_t start = base_idx; - for (; base_idx < next; base_idx++) { - if (ValidityMask::RowIsValid(validity_entry, base_idx - start)) { - auto lentry = ldata[LEFT_CONSTANT ? 0 : base_idx]; - auto rentry = rdata[RIGHT_CONSTANT ? 0 : base_idx]; - result_data[base_idx] = - OPWRAPPER::template Operation( - fun, lentry, rentry, mask, base_idx); - } - } - } - } - } else { - for (idx_t i = 0; i < count; i++) { - auto lentry = ldata[LEFT_CONSTANT ? 0 : i]; - auto rentry = rdata[RIGHT_CONSTANT ? 0 : i]; - result_data[i] = OPWRAPPER::template Operation( - fun, lentry, rentry, mask, i); - } - } + explicit BinaryScalarAdapter(FUNC &fun_p) : fun(fun_p) { } -#endif - template - static void ExecuteConstant(const Vector &left, const Vector &right, Vector &result, idx_t count, FUNC fun) { - result.SetVectorType(VectorType::CONSTANT_VECTOR); - if (result.size() != count) { - FlatVector::SetSize(result, count); - } + inline RESULT_TYPE Operation(ValidityMask &mask, idx_t idx, LEFT_TYPE left, RIGHT_TYPE right) { + return OPWRAPPER::template Operation(fun, left, right, mask, idx); + } - auto ldata = ConstantVector::GetData(left); - auto rdata = ConstantVector::GetData(right); - auto result_data = ConstantVector::GetData(result); + FUNC &fun; +}; - if (ConstantVector::IsNull(left) || ConstantVector::IsNull(right)) { - ConstantVector::SetNull(result, count_t(count)); - return; - } - *result_data = OPWRAPPER::template Operation( - fun, *ldata, *rdata, ConstantVector::Validity(result), 0); +template +struct BinarySelectAdapter { + inline bool Operation(LEFT_TYPE left, RIGHT_TYPE right) { + return OP::Operation(left, right); } -#if !DUCKDB_SMALLER_BINARY(binary_executor_flat) - template - static void ExecuteFlat(const Vector &left, const Vector &right, Vector &result, idx_t count, FUNC fun) { - auto ldata = LEFT_CONSTANT ? ConstantVector::GetData(left) : FlatVector::GetData(left); - auto rdata = - RIGHT_CONSTANT ? ConstantVector::GetData(right) : FlatVector::GetData(right); - - if ((LEFT_CONSTANT && ConstantVector::IsNull(left)) || (RIGHT_CONSTANT && ConstantVector::IsNull(right))) { - // either left or right is constant NULL: result is constant NULL - ConstantVector::SetNull(result, count_t(count)); - return; - } - - result.SetVectorType(VectorType::FLAT_VECTOR); - if (result.size() != count) { - FlatVector::SetSize(result, count); - } - auto result_data = FlatVector::GetDataMutable(result); - auto &result_validity = FlatVector::ValidityMutable(result); - if (LEFT_CONSTANT) { - if (OPWRAPPER::AddsNulls()) { - result_validity.Copy(FlatVector::Validity(right), count); - } else { - FlatVector::SetValidity(result, FlatVector::Validity(right)); - } - } else if (RIGHT_CONSTANT) { - if (OPWRAPPER::AddsNulls()) { - result_validity.Copy(FlatVector::Validity(left), count); - } else { - FlatVector::SetValidity(result, FlatVector::Validity(left)); - } - } else { - if (OPWRAPPER::AddsNulls()) { - result_validity.Copy(FlatVector::Validity(left), count); - if (result_validity.CannotHaveNull()) { - result_validity.Copy(FlatVector::Validity(right), count); - } else { - result_validity.Combine(FlatVector::Validity(right), count); - } - } else { - FlatVector::SetValidity(result, FlatVector::Validity(left)); - result_validity.Combine(FlatVector::Validity(right), count); + inline bool OperationNoNull(LEFT_TYPE left, RIGHT_TYPE right) { + if constexpr (ComparisonSelectComplement::FOLD) { + using FOLDED = ComparisonSelectComplement; + using COMPLEMENT = typename FOLDED::COMPLEMENT; + if constexpr (FOLDED::SWAP_OPERANDS) { + return !COMPLEMENT::Operation(right, left); } + return !COMPLEMENT::Operation(left, right); } - ExecuteFlatLoop( - ldata, rdata, result_data, count, result_validity, fun); + return Operation(left, right); } +}; + +struct BinaryExecutor { +private: + struct ExecutePolicy { +#if !DUCKDB_SMALLER_BINARY(binary_executor_flat) + static constexpr bool SPECIALIZE_FLAT = true; +#else + static constexpr bool SPECIALIZE_FLAT = false; +#endif +#if !DUCKDB_SMALLER_BINARY(binary_executor_generic_nullable) + static constexpr bool SPECIALIZE_NULLABLE_GENERIC_SELECTIONS = true; +#else + static constexpr bool SPECIALIZE_NULLABLE_GENERIC_SELECTIONS = false; +#endif + static constexpr bool PRESERVE_RESULT_VALIDITY = false; + }; + + struct SelectPolicy { +#if !DUCKDB_SMALLER_BINARY(binary_executor_select_flat) + static constexpr uint64_t SPECIALIZED_MASKS = 0x7; + static constexpr uint64_t DIRECT_TRUE_FLAT_MASKS = 0x7; +#else + static constexpr uint64_t SPECIALIZED_MASKS = 0; + static constexpr uint64_t DIRECT_TRUE_FLAT_MASKS = 0; +#endif +#if !DUCKDB_SMALLER_BINARY(binary_executor_select_flags) + static constexpr bool SPECIALIZE_OUTPUTS = true; +#else + static constexpr bool SPECIALIZE_OUTPUTS = false; #endif + }; + + template + static void ExecuteSwitchInternal(const Vector &left, const Vector &right, Vector &result, idx_t count, FUNC &fun) { + std::array inputs = {{left, right}}; + BinaryScalarAdapter adapter(fun); + ScalarExecutor::Execute(inputs, result, + count, adapter); + } template - static void ExecuteGenericLoop(const LEFT_TYPE *__restrict ldata, const RIGHT_TYPE *__restrict rdata, - RESULT_TYPE *__restrict result_data, const SelectionVector *__restrict lsel, - const SelectionVector *__restrict rsel, idx_t count, const ValidityMask &lvalidity, - const ValidityMask &rvalidity, ValidityMask &result_validity, FUNC fun) { - if (lvalidity.CanHaveNull() || rvalidity.CanHaveNull()) { - for (idx_t i = 0; i < count; i++) { - auto lindex = lsel->get_index(i); - auto rindex = rsel->get_index(i); - if (lvalidity.RowIsValid(lindex) && rvalidity.RowIsValid(rindex)) { - auto lentry = ldata[lindex]; - auto rentry = rdata[rindex]; - result_data[i] = OPWRAPPER::template Operation( - fun, lentry, rentry, result_validity, i); - } else { - result_validity.SetInvalid(i); - } - } + static void ExecuteSwitch(const Vector &left, const Vector &right, Vector &result, idx_t count, FUNC &fun) { + if (OPWRAPPER::AddsNulls()) { + ExecuteSwitchInternal(left, right, result, count, + fun); } else { - for (idx_t i = 0; i < count; i++) { - auto lentry = ldata[lsel->get_index(i)]; - auto rentry = rdata[rsel->get_index(i)]; - result_data[i] = OPWRAPPER::template Operation( - fun, lentry, rentry, result_validity, i); - } + ExecuteSwitchInternal(left, right, result, count, + fun); } } - template - static void ExecuteGeneric(const Vector &left, const Vector &right, Vector &result, idx_t count, FUNC fun) { - UnifiedVectorFormat ldata, rdata; - - left.ToUnifiedFormat(ldata); - right.ToUnifiedFormat(rdata); - - result.SetVectorType(VectorType::FLAT_VECTOR); - if (result.size() != count) { - FlatVector::SetSize(result, count); + static idx_t CheckExecuteCount(const Vector &left, const Vector &right) { + if (left.size() != right.size()) { + throw InternalException( + "Mismatch in input vector sizes for BinaryExecutor - left has %d rows but right has %d", left.size(), + right.size()); } - auto result_data = FlatVector::GetDataMutable(result); - ExecuteGenericLoop( - UnifiedVectorFormat::GetData(ldata), UnifiedVectorFormat::GetData(rdata), - result_data, ldata.sel, rdata.sel, count, ldata.validity, rdata.validity, - FlatVector::ValidityMutable(result), fun); + return left.size(); } - template - static void ExecuteSwitch(const Vector &left, const Vector &right, Vector &result, idx_t count, FUNC fun) { - auto left_vector_type = left.GetVectorType(); - auto right_vector_type = right.GetVectorType(); - if (left_vector_type == VectorType::CONSTANT_VECTOR && right_vector_type == VectorType::CONSTANT_VECTOR) { - ExecuteConstant(left, right, result, count, fun); -#if !DUCKDB_SMALLER_BINARY(binary_executor_flat) - } else if (left_vector_type == VectorType::FLAT_VECTOR && right_vector_type == VectorType::CONSTANT_VECTOR) { - ExecuteFlat(left, right, result, - count, fun); - } else if (left_vector_type == VectorType::CONSTANT_VECTOR && right_vector_type == VectorType::FLAT_VECTOR) { - ExecuteFlat(left, right, result, - count, fun); - } else if (left_vector_type == VectorType::FLAT_VECTOR && right_vector_type == VectorType::FLAT_VECTOR) { - ExecuteFlat(left, right, result, - count, fun); -#endif - } else { - ExecuteGeneric(left, right, result, count, fun); - } + template + static idx_t SelectShared(const std::array &inputs, const SelectionVector *sel, + idx_t count, SelectionVector *true_sel, SelectionVector *false_sel) { + BinarySelectAdapter adapter; + return ScalarExecutor::Select( + inputs, sel, count, true_sel, false_sel, adapter); } public: template > static void Execute(const Vector &left, const Vector &right, Vector &result, idx_t count, FUNC fun) { - constexpr bool adds_nulls = std::is_same(), std::declval())), - optional>::value; - ExecuteSwitch, bool, FUNC>( - left, right, result, count, fun); + constexpr bool adds_nulls = + std::is_same, optional>::value; + ExecuteSwitch, bool>(left, right, result, + count, fun); } template static void Execute(const Vector &left, const Vector &right, Vector &result, idx_t count) { - ExecuteSwitch(left, right, result, count, false); + bool dummy = false; + ExecuteSwitch(left, right, result, count, dummy); } template static void ExecuteStandard(const Vector &left, const Vector &right, Vector &result, idx_t count) { - ExecuteSwitch(left, right, result, - count, false); + bool dummy = false; + ExecuteSwitch(left, right, result, count, + dummy); } -private: - static idx_t CheckExecuteCount(const Vector &left, const Vector &right) { - if (left.size() != right.size()) { - throw InternalException( - "Mismatch in input vector sizes for BinaryExecutor - left has %d rows but right has %d", left.size(), - right.size()); - } - return left.size(); - } - -public: - //! Convenience overloads without explicit count - count is derived from the input vectors. template > static void Execute(const Vector &left, const Vector &right, Vector &result, FUNC fun) { @@ -337,408 +230,11 @@ struct BinaryExecutor { ExecuteStandard(left, right, result, CheckExecuteCount(left, right)); } -public: - template - static idx_t SelectConstant(const Vector &left, const Vector &right, const SelectionVector &sel, idx_t count, - SelectionVector *true_sel, SelectionVector *false_sel) { - auto ldata = ConstantVector::GetData(left); - auto rdata = ConstantVector::GetData(right); - - // both sides are constant, return either 0 or the count - // in this case we do not fill in the result selection vector at all - if (ConstantVector::IsNull(left) || ConstantVector::IsNull(right) || !OP::Operation(*ldata, *rdata)) { - if (false_sel) { - for (idx_t i = 0; i < count; i++) { - false_sel->set_index(i, sel.get_index(i)); - } - } - return 0; - } else { - if (true_sel) { - for (idx_t i = 0; i < count; i++) { - true_sel->set_index(i, sel.get_index(i)); - } - } - return count; - } - } - -// NOTE: the flat path is intentionally NOT covered by the ComparisonSelectComplement fold (unlike -// the generic and constant paths above). It is null-unified — there is no NO_NULL template split; -#if !DUCKDB_SMALLER_BINARY(binary_executor_select_flat) - template - static inline idx_t SelectFlatLoop(const LEFT_TYPE *__restrict ldata, const RIGHT_TYPE *__restrict rdata, - const SelectionVector &sel, idx_t count, const ValidityMask &validity_mask, - SelectionVector *true_sel, SelectionVector *false_sel) { - idx_t true_count = 0, false_count = 0; - idx_t base_idx = 0; - auto entry_count = ValidityMask::EntryCount(count); - for (idx_t entry_idx = 0; entry_idx < entry_count; entry_idx++) { - auto validity_entry = validity_mask.GetValidityEntry(entry_idx); - idx_t next = MinValue(base_idx + ValidityMask::BITS_PER_VALUE, count); - if (ValidityMask::AllValid(validity_entry)) { - // all valid: perform operation - for (; base_idx < next; base_idx++) { - idx_t result_idx = sel.get_index(base_idx); - idx_t lidx = LEFT_CONSTANT ? 0 : base_idx; - idx_t ridx = RIGHT_CONSTANT ? 0 : base_idx; - bool comparison_result = OP::Operation(ldata[lidx], rdata[ridx]); - if (HAS_TRUE_SEL) { - true_sel->set_index(true_count, result_idx); - true_count += comparison_result; - } - if (HAS_FALSE_SEL) { - false_sel->set_index(false_count, result_idx); - false_count += !comparison_result; - } - } - } else if (ValidityMask::NoneValid(validity_entry)) { - // nothing valid: skip all - if (HAS_FALSE_SEL) { - for (; base_idx < next; base_idx++) { - idx_t result_idx = sel.get_index(base_idx); - false_sel->set_index(false_count, result_idx); - false_count++; - } - } - base_idx = next; - continue; - } else { - // partially valid: need to check individual elements for validity - idx_t start = base_idx; - for (; base_idx < next; base_idx++) { - idx_t result_idx = sel.get_index(base_idx); - idx_t lidx = LEFT_CONSTANT ? 0 : base_idx; - idx_t ridx = RIGHT_CONSTANT ? 0 : base_idx; - bool comparison_result = ValidityMask::RowIsValid(validity_entry, base_idx - start) && - OP::Operation(ldata[lidx], rdata[ridx]); - if (HAS_TRUE_SEL) { - true_sel->set_index(true_count, result_idx); - true_count += comparison_result; - } - if (HAS_FALSE_SEL) { - false_sel->set_index(false_count, result_idx); - false_count += !comparison_result; - } - } - } - } - if (HAS_TRUE_SEL) { - return true_count; - } else { - return count - false_count; - } - } - - template - static inline idx_t SelectFlatLoopSwitch(const LEFT_TYPE *__restrict ldata, const RIGHT_TYPE *__restrict rdata, - const SelectionVector &sel, idx_t count, const ValidityMask &mask, - SelectionVector *true_sel, SelectionVector *false_sel) { - if (true_sel && false_sel) { - return SelectFlatLoop( - ldata, rdata, sel, count, mask, true_sel, false_sel); - } else if (true_sel) { - return SelectFlatLoop( - ldata, rdata, sel, count, mask, true_sel, false_sel); - } else { - D_ASSERT(false_sel); - return SelectFlatLoop( - ldata, rdata, sel, count, mask, true_sel, false_sel); - } - } - - template - static idx_t SelectFlat(const Vector &left, const Vector &right, const SelectionVector &sel, idx_t count, - SelectionVector *true_sel, SelectionVector *false_sel) { - auto ldata = LEFT_CONSTANT ? ConstantVector::GetData(left) : FlatVector::GetData(left); - auto rdata = - RIGHT_CONSTANT ? ConstantVector::GetData(right) : FlatVector::GetData(right); - - if (LEFT_CONSTANT && ConstantVector::IsNull(left)) { - if (false_sel) { - for (idx_t i = 0; i < count; i++) { - false_sel->set_index(i, sel.get_index(i)); - } - } - return 0; - } - if (RIGHT_CONSTANT && ConstantVector::IsNull(right)) { - if (false_sel) { - for (idx_t i = 0; i < count; i++) { - false_sel->set_index(i, sel.get_index(i)); - } - } - return 0; - } - - if (LEFT_CONSTANT) { - return SelectFlatLoopSwitch( - ldata, rdata, sel, count, FlatVector::Validity(right), true_sel, false_sel); - } else if (RIGHT_CONSTANT) { - return SelectFlatLoopSwitch( - ldata, rdata, sel, count, FlatVector::Validity(left), true_sel, false_sel); - } else { - ValidityMask combined_mask = FlatVector::Validity(left); - combined_mask.Combine(FlatVector::Validity(right), count); - return SelectFlatLoopSwitch( - ldata, rdata, sel, count, combined_mask, true_sel, false_sel); - } - } - - template - static idx_t SelectGenericConstant(CONSTANT_TYPE constant, const GENERIC_TYPE *__restrict data, - const SelectionVector &generic_sel, const ValidityMask &mask, - const SelectionVector &result_sel, idx_t count, SelectionVector *true_sel, - SelectionVector *false_sel) { - idx_t true_count = 0, false_count = 0; - for (idx_t r = 0; r < count; r++) { - auto result_idx = result_sel.get_index(r); - auto idx = generic_sel.get_index(r); - bool comparison_result = (!CAN_HAVE_NULL || mask.RowIsValid(idx)); - if constexpr (RIGHT_CONSTANT) { - comparison_result = comparison_result && OP::Operation(data[idx], constant); - } else { - comparison_result = comparison_result && OP::Operation(constant, data[idx]); - } - if constexpr (HAS_TRUE_SEL) { - true_sel->set_index(true_count, result_idx); - true_count += comparison_result; - } - if constexpr (HAS_FALSE_SEL) { - false_sel->set_index(false_count, result_idx); - false_count += !comparison_result; - } - } - if constexpr (HAS_TRUE_SEL) { - return true_count; - } else { - return count - false_count; - } - } - - template - static idx_t SelectGenericConstant(CONSTANT_TYPE constant, const GENERIC_TYPE *__restrict data, - const SelectionVector &generic_sel, const ValidityMask &mask, - const SelectionVector &result_sel, idx_t count, SelectionVector *true_sel, - SelectionVector *false_sel) { - // NO_NULL complement fold: NotEquals/GreaterThanEquals are exactly the complement of - // Equals/GreaterThan when there are no NULLs - if constexpr (!CAN_HAVE_NULL && ComparisonSelectComplement::FOLD) { - using FOLDED = ComparisonSelectComplement; - constexpr bool FOLDED_RIGHT_CONSTANT = FOLDED::SWAP_OPERANDS ? !RIGHT_CONSTANT : RIGHT_CONSTANT; - return count - SelectGenericConstant( - constant, data, generic_sel, mask, result_sel, count, false_sel, true_sel); - } - if (true_sel && false_sel) { - return SelectGenericConstant( - constant, data, generic_sel, mask, result_sel, count, true_sel, false_sel); - } else if (true_sel) { - return SelectGenericConstant( - constant, data, generic_sel, mask, result_sel, count, true_sel, false_sel); - } else if (false_sel) { - return SelectGenericConstant( - constant, data, generic_sel, mask, result_sel, count, true_sel, false_sel); - } else { - throw InternalException("Either true or false sel must be set"); - } - } - - template - static idx_t SelectGenericConstant(CONSTANT_TYPE constant, const GENERIC_TYPE *__restrict data, - const SelectionVector &generic_sel, const ValidityMask &mask, - const SelectionVector &sel, idx_t count, SelectionVector *true_sel, - SelectionVector *false_sel) { - if (mask.CanHaveNull()) { - return SelectGenericConstant( - constant, data, generic_sel, mask, sel, count, true_sel, false_sel); - } else { - return SelectGenericConstant( - constant, data, generic_sel, mask, sel, count, true_sel, false_sel); - } - } - - template - static idx_t SelectGenericConstant(const Vector &left, const Vector &right, const SelectionVector &sel, idx_t count, - SelectionVector *true_sel, SelectionVector *false_sel) { - constexpr bool LEFT_CONSTANT = !RIGHT_CONSTANT; - if (LEFT_CONSTANT && ConstantVector::IsNull(left)) { - if (false_sel) { - for (idx_t i = 0; i < count; i++) { - false_sel->set_index(i, sel.get_index(i)); - } - } - return 0; - } - if (RIGHT_CONSTANT && ConstantVector::IsNull(right)) { - if (false_sel) { - for (idx_t i = 0; i < count; i++) { - false_sel->set_index(i, sel.get_index(i)); - } - } - return 0; - } - - UnifiedVectorFormat format; - if (LEFT_CONSTANT) { - } else { - right.ToUnifiedFormat(format); - } - - if (LEFT_CONSTANT) { - right.ToUnifiedFormat(format); - auto data = UnifiedVectorFormat::GetData(format); - return SelectGenericConstant( - *ConstantVector::GetData(left), data, *format.sel, format.validity, sel, count, true_sel, - false_sel); - } else { - left.ToUnifiedFormat(format); - auto data = UnifiedVectorFormat::GetData(format); - return SelectGenericConstant( - *ConstantVector::GetData(right), data, *format.sel, format.validity, sel, count, true_sel, - false_sel); - } - } -#endif - -#if !DUCKDB_SMALLER_BINARY(binary_executor_select_flags) - template -#else - template -#endif - static inline idx_t SelectGenericLoop(const LEFT_TYPE *__restrict ldata, const RIGHT_TYPE *__restrict rdata, - const SelectionVector *__restrict lsel, - const SelectionVector *__restrict rsel, const SelectionVector &result_sel, - idx_t count, const ValidityMask &lvalidity, const ValidityMask &rvalidity, - SelectionVector *true_sel, SelectionVector *false_sel) { - idx_t true_count = 0, false_count = 0; -#if DUCKDB_SMALLER_BINARY(binary_executor_select_flags) - const bool HAS_TRUE_SEL = true_sel; - const bool HAS_FALSE_SEL = false_sel; - const bool NO_NULL = false; -#endif - for (idx_t i = 0; i < count; i++) { - auto result_idx = result_sel.get_index(i); - auto lindex = lsel->get_index(i); - auto rindex = rsel->get_index(i); - const bool comparison_result = - (NO_NULL || (lvalidity.RowIsValid(lindex) && rvalidity.RowIsValid(rindex))) && - OP::Operation(ldata[lindex], rdata[rindex]); - if (HAS_TRUE_SEL) { - true_sel->set_index(true_count, result_idx); - true_count += comparison_result; - } - if (HAS_FALSE_SEL) { - false_sel->set_index(false_count, result_idx); - false_count += !comparison_result; - } - } - if (HAS_TRUE_SEL) { - return true_count; - } else { - return count - false_count; - } - } - -#if !DUCKDB_SMALLER_BINARY(binary_executor_select_flags) - template - static inline idx_t - SelectGenericLoopSelSwitch(const LEFT_TYPE *__restrict ldata, const RIGHT_TYPE *__restrict rdata, - const SelectionVector *__restrict lsel, const SelectionVector *__restrict rsel, - const SelectionVector &result_sel, idx_t count, const ValidityMask &lvalidity, - const ValidityMask &rvalidity, SelectionVector *true_sel, SelectionVector *false_sel) { - // NO_NULL complement fold: NotEquals/GreaterThanEquals are exactly the complement of - // Equals/GreaterThan when there are no NULLs - if constexpr (NO_NULL && ComparisonSelectComplement::FOLD) { - using FOLDED = ComparisonSelectComplement; - if constexpr (FOLDED::SWAP_OPERANDS) { - return count - - SelectGenericLoopSelSwitch( - rdata, ldata, rsel, lsel, result_sel, count, rvalidity, lvalidity, false_sel, true_sel); - } else { - return count - - SelectGenericLoopSelSwitch( - ldata, rdata, lsel, rsel, result_sel, count, lvalidity, rvalidity, false_sel, true_sel); - } - } - if (true_sel && false_sel) { - return SelectGenericLoop( - ldata, rdata, lsel, rsel, result_sel, count, lvalidity, rvalidity, true_sel, false_sel); - } else if (true_sel) { - return SelectGenericLoop( - ldata, rdata, lsel, rsel, result_sel, count, lvalidity, rvalidity, true_sel, false_sel); - } else { - D_ASSERT(false_sel); - return SelectGenericLoop( - ldata, rdata, lsel, rsel, result_sel, count, lvalidity, rvalidity, true_sel, false_sel); - } - } -#endif - - template - static inline idx_t - SelectGenericLoopSwitch(const LEFT_TYPE *__restrict ldata, const RIGHT_TYPE *__restrict rdata, - const SelectionVector *__restrict lsel, const SelectionVector *__restrict rsel, - const SelectionVector &result_sel, idx_t count, const ValidityMask &lvalidity, - const ValidityMask &rvalidity, SelectionVector *true_sel, SelectionVector *false_sel) { -#if !DUCKDB_SMALLER_BINARY(binary_executor_select_flags) - if (lvalidity.CanHaveNull() || rvalidity.CanHaveNull()) { - return SelectGenericLoopSelSwitch( - ldata, rdata, lsel, rsel, result_sel, count, lvalidity, rvalidity, true_sel, false_sel); - } else { - return SelectGenericLoopSelSwitch( - ldata, rdata, lsel, rsel, result_sel, count, lvalidity, rvalidity, true_sel, false_sel); - } -#else - return SelectGenericLoop(ldata, rdata, lsel, rsel, result_sel, count, lvalidity, - rvalidity, true_sel, false_sel); -#endif - } - - template - static idx_t SelectGeneric(const Vector &left, const Vector &right, const SelectionVector &sel, idx_t count, - SelectionVector *true_sel, SelectionVector *false_sel) { - UnifiedVectorFormat ldata, rdata; - - left.ToUnifiedFormat(ldata); - right.ToUnifiedFormat(rdata); - - return SelectGenericLoopSwitch( - UnifiedVectorFormat::GetData(ldata), UnifiedVectorFormat::GetData(rdata), ldata.sel, - rdata.sel, sel, count, ldata.validity, rdata.validity, true_sel, false_sel); - } - template static idx_t Select(const Vector &left, const Vector &right, const SelectionVector *sel, idx_t count, SelectionVector *true_sel, SelectionVector *false_sel) { - if (!sel) { - sel = FlatVector::IncrementalSelectionVector(); - } - if (left.GetVectorType() == VectorType::CONSTANT_VECTOR && - right.GetVectorType() == VectorType::CONSTANT_VECTOR) { - return SelectConstant(left, right, *sel, count, true_sel, false_sel); -#if !DUCKDB_SMALLER_BINARY(binary_executor_select_flat) - } else if (left.GetVectorType() == VectorType::CONSTANT_VECTOR && - right.GetVectorType() == VectorType::FLAT_VECTOR) { - return SelectFlat(left, right, *sel, count, true_sel, false_sel); - } else if (left.GetVectorType() == VectorType::FLAT_VECTOR && - right.GetVectorType() == VectorType::CONSTANT_VECTOR) { - return SelectFlat(left, right, *sel, count, true_sel, false_sel); - } else if (left.GetVectorType() == VectorType::FLAT_VECTOR && - right.GetVectorType() == VectorType::FLAT_VECTOR) { - return SelectFlat(left, right, *sel, count, true_sel, false_sel); - } else if (left.GetVectorType() == VectorType::CONSTANT_VECTOR) { - return SelectGenericConstant(left, right, *sel, count, true_sel, - false_sel); - } else if (right.GetVectorType() == VectorType::CONSTANT_VECTOR) { - return SelectGenericConstant(left, right, *sel, count, true_sel, - false_sel); -#endif - } else { - return SelectGeneric(left, right, *sel, count, true_sel, false_sel); - } + std::array inputs = {{left, right}}; + return SelectShared(inputs, sel, count, true_sel, false_sel); } }; diff --git a/src/duckdb/src/include/duckdb/common/vector_operations/scalar_executor.hpp b/src/duckdb/src/include/duckdb/common/vector_operations/scalar_executor.hpp new file mode 100644 index 000000000..26d404376 --- /dev/null +++ b/src/duckdb/src/include/duckdb/common/vector_operations/scalar_executor.hpp @@ -0,0 +1,813 @@ +//===----------------------------------------------------------------------===// +// DuckDB +// +// duckdb/common/vector_operations/scalar_executor.hpp +// +//===----------------------------------------------------------------------===// + +#pragma once + +#include "duckdb/common/exception.hpp" +#include "duckdb/common/types/vector.hpp" +#include "duckdb/common/vector/constant_vector.hpp" +#include "duckdb/common/vector/flat_vector.hpp" + +#include +#include +#include +#include + +#if defined(_MSC_VER) +#define DUCKDB_SCALAR_EXECUTOR_NOINLINE __declspec(noinline) +#elif defined(__GNUC__) +#define DUCKDB_SCALAR_EXECUTOR_NOINLINE __attribute__((noinline)) +#else +#define DUCKDB_SCALAR_EXECUTOR_NOINLINE +#endif + +namespace duckdb { + +//! Internal execution engine shared by the named scalar executor facades. +struct ScalarExecutor { + using VectorRef = std::reference_wrapper; + +private: + struct InputProfile { + uint64_t constant_mask = 0; + bool all_constant = true; + bool all_flat_or_constant = true; + bool any_constant_null = false; + }; + + template + static inline InputProfile GetInputProfile(const std::array &inputs, std::index_sequence) { + InputProfile result; + auto classify = [&](auto input_index) { + constexpr idx_t INPUT_INDEX = decltype(input_index)::value; + auto vector_type = inputs[INPUT_INDEX].get().GetVectorType(); + if (vector_type == VectorType::CONSTANT_VECTOR) { + if constexpr (INPUT_INDEX < 64) { + result.constant_mask |= uint64_t(1) << INPUT_INDEX; + } + result.any_constant_null |= ConstantVector::IsNull(inputs[INPUT_INDEX].get()); + } else { + result.all_constant = false; + result.all_flat_or_constant &= vector_type == VectorType::FLAT_VECTOR; + } + }; + (classify(std::integral_constant {}), ...); + return result; + } + + template + static inline idx_t InputIndex(idx_t row) { + if constexpr (CONSTANT_MASK & (uint64_t(1) << INPUT_INDEX)) { + return 0; + } else { + return row; + } + } + + template + static inline ValidityMask &PrepareFlatResultValidity(const std::array &inputs, Vector &result, + idx_t count, std::index_sequence) { + auto &result_validity = FlatVector::ValidityMutable(result); + bool initialized = false; + auto combine = [&](auto input_index) { + constexpr idx_t INPUT_INDEX = decltype(input_index)::value; + if constexpr (!(CONSTANT_MASK & (uint64_t(1) << INPUT_INDEX))) { + auto &input_validity = FlatVector::Validity(inputs[INPUT_INDEX].get()); + if (input_validity.CanHaveNull()) { + if (!initialized) { + if constexpr (ADDS_NULLS) { + result_validity.Copy(input_validity, count); + } else { + result_validity.Initialize(input_validity); + } + initialized = true; + } else { + result_validity.Combine(input_validity, count); + } + } + } + }; + (combine(std::integral_constant {}), ...); + if (!initialized) { + if constexpr (PRESERVE_RESULT_VALIDITY) { + if constexpr (ADDS_NULLS) { + ValidityMask preserved(result_validity, count); + result_validity.Initialize(preserved); + } + } else { + result_validity.Reset(count); + } + } + return result_validity; + } + + template + static inline ValidityMask PrepareFlatInputValidity(const std::array &inputs, idx_t count, + std::index_sequence) { + ValidityMask result(count); + bool initialized = false; + auto combine = [&](auto input_index) { + constexpr idx_t INPUT_INDEX = decltype(input_index)::value; + if constexpr (!(CONSTANT_MASK & (uint64_t(1) << INPUT_INDEX))) { + auto &input_validity = FlatVector::Validity(inputs[INPUT_INDEX].get()); + if (!initialized) { + result.Initialize(input_validity); + initialized = true; + } else { + result.Combine(input_validity, count); + } + } + }; + (combine(std::integral_constant {}), ...); + return result; + } + + DUCKDB_API static bool PrepareGenericResultValidity(const UnifiedVectorFormat *formats, idx_t format_count, + Vector &result, idx_t count, bool preserve_result_validity, + bool adds_nulls); + + template + static void ExecuteFlat(const std::array &inputs, Vector &result, idx_t count, + ADAPTER &adapter, std::index_sequence indices) { + result.SetVectorType(VectorType::FLAT_VECTOR); + if (result.size() != count) { + FlatVector::SetSize(result, count); + } + auto result_data = FlatVector::GetDataMutable(result); + auto input_data = std::make_tuple(FlatVector::GetData(inputs[Is].get())...); + auto &result_validity = PrepareFlatResultValidity( + inputs, result, count, indices); + +#ifdef DEBUG + auto assert_restrict = [&](auto input_index) { + constexpr idx_t INPUT_INDEX = decltype(input_index)::value; + if constexpr (!(CONSTANT_MASK & (uint64_t(1) << INPUT_INDEX))) { + auto data = std::get(input_data); + ASSERT_RESTRICT(data, data + count, result_data, result_data + count); + } + }; + (assert_restrict(std::integral_constant {}), ...); +#endif + + if (result_validity.CanHaveNull()) { + idx_t base_idx = 0; + auto entry_count = ValidityMask::EntryCount(count); + for (idx_t entry_idx = 0; entry_idx < entry_count; entry_idx++) { + auto validity_entry = result_validity.GetValidityEntry(entry_idx); + auto next = MinValue(base_idx + ValidityMask::BITS_PER_VALUE, count); + if (ValidityMask::AllValid(validity_entry)) { + for (; base_idx < next; base_idx++) { + result_data[base_idx] = + adapter.Operation(result_validity, base_idx, + std::get(input_data)[InputIndex(base_idx)]...); + } + } else if (ValidityMask::NoneValid(validity_entry)) { + base_idx = next; + } else { + auto start = base_idx; + for (; base_idx < next; base_idx++) { + if (ValidityMask::RowIsValid(validity_entry, base_idx - start)) { + result_data[base_idx] = + adapter.Operation(result_validity, base_idx, + std::get(input_data)[InputIndex(base_idx)]...); + } + } + } + } + } else { + for (idx_t row = 0; row < count; row++) { + result_data[row] = adapter.Operation(result_validity, row, + std::get(input_data)[InputIndex(row)]...); + } + } + } + + template + static void ExecuteConstant(const std::array &inputs, Vector &result, idx_t count, + bool input_is_null, ADAPTER &adapter, std::index_sequence) { + result.SetVectorType(VectorType::CONSTANT_VECTOR); + if (result.size() != count) { + FlatVector::SetSize(result, count); + } + if (input_is_null) { + ConstantVector::SetNull(result, true); + return; + } + auto &result_validity = ConstantVector::Validity(result); + result_validity.SetValid(0); + auto result_data = ConstantVector::GetData(result); + result_data[0] = adapter.Operation(result_validity, 0, *ConstantVector::GetData(inputs[Is].get())...); + } + + template + static inline idx_t GenericInputIndex(const std::array &selections, idx_t row) { + if constexpr (SELECTION_MASK & (uint64_t(1) << INPUT_INDEX)) { + return selections[INPUT_INDEX][row]; + } + return row; + } + + template + static void + ExecuteGenericBinaryNullable(const LEFT_TYPE *__restrict left_data, const RIGHT_TYPE *__restrict right_data, + const std::array &formats, + const std::array &selections, RESULT_TYPE *__restrict result_data, + idx_t count, ValidityMask &result_validity, ADAPTER &adapter) { + auto &left_validity = formats[0].validity; + auto &right_validity = formats[1].validity; + for (idx_t row = 0; row < count; row++) { + auto left_index = GenericInputIndex<0, SELECTION_MASK>(selections, row); + auto right_index = GenericInputIndex<1, SELECTION_MASK>(selections, row); + if (left_validity.RowIsValid(left_index) && right_validity.RowIsValid(right_index)) { + result_data[row] = + adapter.Operation(result_validity, row, left_data[left_index], right_data[right_index]); + } else { + result_validity.SetInvalid(row); + } + } + } + + template + static void ExecuteGenericBinaryNullableSwitch(const LEFT_TYPE *__restrict left_data, + const RIGHT_TYPE *__restrict right_data, + const std::array &formats, + RESULT_TYPE *__restrict result_data, idx_t count, + ValidityMask &result_validity, ADAPTER &adapter) { + std::array selections = {{formats[0].sel->data(), formats[1].sel->data()}}; + uint64_t selection_mask = 0; + selection_mask |= selections[0] ? 1 : 0; + selection_mask |= selections[1] ? 2 : 0; + switch (selection_mask) { + case 0: + ExecuteGenericBinaryNullable<0>(left_data, right_data, formats, selections, result_data, count, + result_validity, adapter); + return; + case 1: + ExecuteGenericBinaryNullable<1>(left_data, right_data, formats, selections, result_data, count, + result_validity, adapter); + return; + case 2: + ExecuteGenericBinaryNullable<2>(left_data, right_data, formats, selections, result_data, count, + result_validity, adapter); + return; + case 3: + ExecuteGenericBinaryNullable<3>(left_data, right_data, formats, selections, result_data, count, + result_validity, adapter); + return; + default: + throw InternalException("Invalid nullable generic scalar executor selection profile"); + } + } + + template + static void ExecuteGeneric(const std::array &inputs, Vector &result, idx_t count, + ADAPTER &adapter, std::index_sequence) { + constexpr idx_t N = sizeof...(ARGS); + std::array formats; + for (idx_t i = 0; i < N; i++) { + inputs[i].get().ToUnifiedFormat(formats[i]); + } + auto input_data = std::make_tuple(UnifiedVectorFormat::GetData(formats[Is])...); + result.SetVectorType(VectorType::FLAT_VECTOR); + if (result.size() != count) { + FlatVector::SetSize(result, count); + } + auto result_data = FlatVector::GetDataMutable(result); + auto &result_validity = FlatVector::ValidityMutable(result); + auto inputs_can_have_null = PrepareGenericResultValidity(formats.data(), N, result, count, + PRESERVE_RESULT_VALIDITY, ADAPTER::ADDS_NULLS); + + if (inputs_can_have_null) { + if constexpr (SPECIALIZE_NULLABLE_GENERIC_SELECTIONS && N == 2) { + ExecuteGenericBinaryNullableSwitch(std::get<0>(input_data), std::get<1>(input_data), formats, + result_data, count, result_validity, adapter); + } else { + for (idx_t row = 0; row < count; row++) { + std::array input_indices = {{formats[Is].sel->get_index(row)...}}; + if ((... && formats[Is].validity.RowIsValid(input_indices[Is]))) { + result_data[row] = + adapter.Operation(result_validity, row, std::get(input_data)[input_indices[Is]]...); + } else { + result_validity.SetInvalid(row); + } + } + } + } else { + for (idx_t row = 0; row < count; row++) { + result_data[row] = adapter.Operation(result_validity, row, + std::get(input_data)[formats[Is].sel->get_index(row)]...); + } + } + } + + template + static void ExecuteInternal(const std::array &inputs, Vector &result, idx_t count, + ADAPTER &adapter, std::index_sequence indices) { + constexpr idx_t N = sizeof...(ARGS); + static_assert(N > 0, "ScalarExecutor requires at least one input"); + auto profile = GetInputProfile(inputs, indices); + if (profile.all_constant) { + ExecuteConstant(inputs, result, count, profile.any_constant_null, adapter, + indices); + return; + } + if constexpr (POLICY::SPECIALIZE_FLAT && N <= 3) { + if (profile.all_flat_or_constant) { + if (profile.any_constant_null) { + result.SetVectorType(VectorType::CONSTANT_VECTOR); + if (result.size() != count) { + FlatVector::SetSize(result, count); + } + ConstantVector::SetNull(result, true); + return; + } + switch (profile.constant_mask) { + case 0: + ExecuteFlat( + inputs, result, count, adapter, indices); + return; + case 1: + if constexpr (N >= 2) { + ExecuteFlat( + inputs, result, count, adapter, indices); + return; + } + break; + case 2: + if constexpr (N >= 2) { + ExecuteFlat( + inputs, result, count, adapter, indices); + return; + } + break; + case 3: + if constexpr (N >= 3) { + ExecuteFlat( + inputs, result, count, adapter, indices); + return; + } + break; + case 4: + if constexpr (N >= 3) { + ExecuteFlat( + inputs, result, count, adapter, indices); + return; + } + break; + case 5: + if constexpr (N >= 3) { + ExecuteFlat( + inputs, result, count, adapter, indices); + return; + } + break; + case 6: + if constexpr (N >= 3) { + ExecuteFlat( + inputs, result, count, adapter, indices); + return; + } + break; + default: + throw InternalException("Invalid flat/constant scalar executor profile"); + } + } + } + ExecuteGeneric(inputs, result, count, adapter, indices); + } + + template + struct StaticSelectionSink { + static_assert(HAS_TRUE_SELECTION || HAS_FALSE_SELECTION, "A selection sink requires an output"); + + StaticSelectionSink(SelectionVector *true_selection_p, SelectionVector *false_selection_p) + : true_selection(true_selection_p ? true_selection_p->data() : nullptr), + false_selection(false_selection_p ? false_selection_p->data() : nullptr) { + D_ASSERT(!HAS_TRUE_SELECTION || true_selection); + D_ASSERT(!HAS_FALSE_SELECTION || false_selection); + } + + inline void Append(bool comparison_result, idx_t result_idx) { + if constexpr (HAS_TRUE_SELECTION) { + true_selection[true_count] = UnsafeNumericCast(result_idx); + true_count += comparison_result; + } + if constexpr (HAS_FALSE_SELECTION) { + false_selection[false_count] = UnsafeNumericCast(result_idx); + false_count += !comparison_result; + } + } + + inline void AppendInvalidRange(const SelectionVector &sel, idx_t start, idx_t end) { + if constexpr (HAS_FALSE_SELECTION) { + for (idx_t row = start; row < end; row++) { + false_selection[false_count++] = UnsafeNumericCast(sel.get_index(row)); + } + } + } + + idx_t FillConstant(bool comparison_result, const SelectionVector &sel, idx_t count) { + if (comparison_result) { + if constexpr (HAS_TRUE_SELECTION) { + for (idx_t row = 0; row < count; row++) { + true_selection[row] = UnsafeNumericCast(sel.get_index(row)); + } + true_count = count; + } + } else if constexpr (HAS_FALSE_SELECTION) { + for (idx_t row = 0; row < count; row++) { + false_selection[row] = UnsafeNumericCast(sel.get_index(row)); + } + false_count = count; + } + return Result(count); + } + + inline idx_t Result(idx_t count) const { + if constexpr (HAS_TRUE_SELECTION) { + return true_count; + } + return count - false_count; + } + + sel_t *true_selection; + sel_t *false_selection; + idx_t true_count = 0; + idx_t false_count = 0; + }; + + struct RuntimeSelectionSink { + RuntimeSelectionSink(SelectionVector *true_selection_p, SelectionVector *false_selection_p) + : true_selection(true_selection_p), false_selection(false_selection_p) { + } + + inline void Append(bool comparison_result, idx_t result_idx) { + if (true_selection) { + true_selection->set_index(true_count, result_idx); + true_count += comparison_result; + } + if (false_selection) { + false_selection->set_index(false_count, result_idx); + false_count += !comparison_result; + } + } + + inline void AppendInvalidRange(const SelectionVector &sel, idx_t start, idx_t end) { + if (false_selection) { + for (idx_t row = start; row < end; row++) { + false_selection->set_index(false_count++, sel.get_index(row)); + } + } + } + + idx_t FillConstant(bool comparison_result, const SelectionVector &sel, idx_t count) { + if (comparison_result) { + if (true_selection) { + for (idx_t row = 0; row < count; row++) { + true_selection->set_index(row, sel.get_index(row)); + } + true_count = count; + } + } else if (false_selection) { + for (idx_t row = 0; row < count; row++) { + false_selection->set_index(row, sel.get_index(row)); + } + false_count = count; + } + return Result(count); + } + + inline idx_t Result(idx_t count) const { + return true_selection ? true_count : count - false_count; + } + + SelectionVector *true_selection; + SelectionVector *false_selection; + idx_t true_count = 0; + idx_t false_count = 0; + }; + + template + static inline bool SelectOperation(ADAPTER &adapter, ARGS... args) { + if constexpr (NO_NULL) { + return adapter.OperationNoNull(args...); + } + return adapter.Operation(args...); + } + + template + static idx_t SelectConstant(const std::array &inputs, const SelectionVector &sel, + idx_t count, const SINK &sink, ADAPTER &adapter, std::index_sequence) { + auto local_sink = sink; + bool comparison_result = adapter.Operation(*ConstantVector::GetData(inputs[Is].get())...); + return local_sink.FillConstant(comparison_result, sel, count); + } + + template + static idx_t SelectFlatLoop(const std::tuple &input_data, const ValidityMask &input_validity, + const SelectionVector &sel, idx_t count, const SINK &sink, ADAPTER &adapter, + std::index_sequence) { + auto local_sink = sink; + idx_t base_idx = 0; + auto entry_count = ValidityMask::EntryCount(count); + for (idx_t entry_idx = 0; entry_idx < entry_count; entry_idx++) { + auto validity_entry = input_validity.GetValidityEntry(entry_idx); + auto next = MinValue(base_idx + ValidityMask::BITS_PER_VALUE, count); + if (ValidityMask::AllValid(validity_entry)) { + for (; base_idx < next; base_idx++) { + auto result_idx = sel.get_index(base_idx); + bool comparison_result = + adapter.Operation(std::get(input_data)[InputIndex(base_idx)]...); + local_sink.Append(comparison_result, result_idx); + } + } else if (ValidityMask::NoneValid(validity_entry)) { + local_sink.AppendInvalidRange(sel, base_idx, next); + base_idx = next; + } else { + auto start = base_idx; + for (; base_idx < next; base_idx++) { + auto result_idx = sel.get_index(base_idx); + bool comparison_result = + ValidityMask::RowIsValid(validity_entry, base_idx - start) && + adapter.Operation(std::get(input_data)[InputIndex(base_idx)]...); + local_sink.Append(comparison_result, result_idx); + } + } + } + return local_sink.Result(count); + } + + template + static idx_t SelectFlat(const std::array &inputs, const SelectionVector &sel, + idx_t count, const SINK &sink, ADAPTER &adapter, std::index_sequence indices) { + auto input_data = std::make_tuple(FlatVector::GetData(inputs[Is].get())...); + auto input_validity = PrepareFlatInputValidity(inputs, count, indices); + return SelectFlatLoop(input_data, input_validity, sel, count, sink, + adapter, indices); + } + + template + static bool TrySelectFlat(const std::array &inputs, const SelectionVector &sel, + idx_t count, uint64_t constant_mask, const SINK &sink, ADAPTER &adapter, + std::index_sequence indices, idx_t &result) { + constexpr idx_t N = sizeof...(ARGS); + switch (constant_mask) { + case 0: + if constexpr (SPECIALIZED_MASKS & (uint64_t(1) << 0)) { + result = SelectFlat<0, SINK, ADAPTER, ARGS...>(inputs, sel, count, sink, adapter, indices); + return true; + } + break; + case 1: + if constexpr (N >= 2 && (SPECIALIZED_MASKS & (uint64_t(1) << 1))) { + result = SelectFlat<1, SINK, ADAPTER, ARGS...>(inputs, sel, count, sink, adapter, indices); + return true; + } + break; + case 2: + if constexpr (N >= 2 && (SPECIALIZED_MASKS & (uint64_t(1) << 2))) { + result = SelectFlat<2, SINK, ADAPTER, ARGS...>(inputs, sel, count, sink, adapter, indices); + return true; + } + break; + case 3: + if constexpr (N >= 3 && (SPECIALIZED_MASKS & (uint64_t(1) << 3))) { + result = SelectFlat<3, SINK, ADAPTER, ARGS...>(inputs, sel, count, sink, adapter, indices); + return true; + } + break; + case 4: + if constexpr (N >= 3 && (SPECIALIZED_MASKS & (uint64_t(1) << 4))) { + result = SelectFlat<4, SINK, ADAPTER, ARGS...>(inputs, sel, count, sink, adapter, indices); + return true; + } + break; + case 5: + if constexpr (N >= 3 && (SPECIALIZED_MASKS & (uint64_t(1) << 5))) { + result = SelectFlat<5, SINK, ADAPTER, ARGS...>(inputs, sel, count, sink, adapter, indices); + return true; + } + break; + case 6: + if constexpr (N >= 3 && (SPECIALIZED_MASKS & (uint64_t(1) << 6))) { + result = SelectFlat<6, SINK, ADAPTER, ARGS...>(inputs, sel, count, sink, adapter, indices); + return true; + } + break; + default: + break; + } + return false; + } + + template + static idx_t SelectGenericConstantLoop(CONSTANT_TYPE constant, const GENERIC_TYPE *__restrict data, + const SelectionVector &generic_sel, const ValidityMask &validity, + const SelectionVector &sel, idx_t count, const SINK &sink, + ADAPTER &adapter) { + auto local_sink = sink; + for (idx_t row = 0; row < count; row++) { + auto result_idx = sel.get_index(row); + auto generic_index = generic_sel.get_index(row); + bool comparison_result = !CAN_HAVE_NULL || validity.RowIsValid(generic_index); + if (comparison_result) { + if constexpr (RIGHT_CONSTANT) { + comparison_result = CAN_HAVE_NULL ? adapter.Operation(data[generic_index], constant) + : adapter.OperationNoNull(data[generic_index], constant); + } else { + comparison_result = CAN_HAVE_NULL ? adapter.Operation(constant, data[generic_index]) + : adapter.OperationNoNull(constant, data[generic_index]); + } + } + local_sink.Append(comparison_result, result_idx); + } + return local_sink.Result(count); + } + + template + static idx_t SelectGenericConstant(const std::array &inputs, const SelectionVector &sel, idx_t count, + const SINK &sink, ADAPTER &adapter) { + static_assert(CONSTANT_MASK == 1 || CONSTANT_MASK == 2, "Exactly one binary input must be constant"); + constexpr idx_t GENERIC_INDEX = CONSTANT_MASK == 1 ? 1 : 0; + UnifiedVectorFormat generic_format; + inputs[GENERIC_INDEX].get().ToUnifiedFormat(generic_format); + auto can_have_null = generic_format.validity.CanHaveNull(); + if constexpr (CONSTANT_MASK == 1) { + auto constant = *ConstantVector::GetData(inputs[0].get()); + auto data = UnifiedVectorFormat::GetData(generic_format); + if (can_have_null) { + return SelectGenericConstantLoop(constant, data, *generic_format.sel, + generic_format.validity, sel, count, sink, adapter); + } + return SelectGenericConstantLoop(constant, data, *generic_format.sel, generic_format.validity, + sel, count, sink, adapter); + } + auto constant = *ConstantVector::GetData(inputs[1].get()); + auto data = UnifiedVectorFormat::GetData(generic_format); + if (can_have_null) { + return SelectGenericConstantLoop(constant, data, *generic_format.sel, generic_format.validity, + sel, count, sink, adapter); + } + return SelectGenericConstantLoop(constant, data, *generic_format.sel, generic_format.validity, sel, + count, sink, adapter); + } + + template + static idx_t SelectGenericLoop(std::tuple &input_data, + std::array &formats, + const SelectionVector &sel, idx_t count, const SINK &sink, ADAPTER &adapter, + std::index_sequence) { + auto local_sink = sink; + constexpr idx_t N = sizeof...(ARGS); + for (idx_t row = 0; row < count; row++) { + auto result_idx = sel.get_index(row); + std::array input_indices = {{formats[Is].sel->get_index(row)...}}; + bool comparison_result = (NO_NULL || (... && formats[Is].validity.RowIsValid(input_indices[Is]))) && + SelectOperation(adapter, std::get(input_data)[input_indices[Is]]...); + local_sink.Append(comparison_result, result_idx); + } + return local_sink.Result(count); + } + + template + static idx_t SelectGeneric(const std::array &inputs, const SelectionVector &sel, + idx_t count, const SINK &sink, ADAPTER &adapter, std::index_sequence indices) { + std::array formats; + for (idx_t i = 0; i < sizeof...(ARGS); i++) { + inputs[i].get().ToUnifiedFormat(formats[i]); + } + auto input_data = std::make_tuple(UnifiedVectorFormat::GetData(formats[Is])...); + if ((... || formats[Is].validity.CanHaveNull())) { + return SelectGenericLoop(input_data, formats, sel, count, sink, adapter, + indices); + } + return SelectGenericLoop(input_data, formats, sel, count, sink, adapter, indices); + } + + template + static idx_t SelectInternal(const std::array &inputs, const SelectionVector &sel, + idx_t count, const InputProfile &profile, const SINK &sink, ADAPTER &adapter, + std::index_sequence indices) { + constexpr idx_t N = sizeof...(ARGS); + if (profile.all_constant) { + if (profile.any_constant_null) { + auto local_sink = sink; + return local_sink.FillConstant(false, sel, count); + } + return SelectConstant(inputs, sel, count, sink, adapter, indices); + } + if constexpr (POLICY::SPECIALIZED_MASKS != 0 && N <= 3) { + if (profile.all_flat_or_constant) { + if (profile.any_constant_null) { + auto local_sink = sink; + return local_sink.FillConstant(false, sel, count); + } + idx_t result; + if (TrySelectFlat( + inputs, sel, count, profile.constant_mask, sink, adapter, indices, result)) { + return result; + } + } + } + if constexpr (POLICY::SPECIALIZED_MASKS != 0 && N == 2) { + if (!profile.any_constant_null) { + switch (profile.constant_mask) { + case 1: + return SelectGenericConstant<1, SINK, ADAPTER, ARGS...>(inputs, sel, count, sink, adapter); + case 2: + return SelectGenericConstant<2, SINK, ADAPTER, ARGS...>(inputs, sel, count, sink, adapter); + default: + break; + } + } + } + return SelectGeneric(inputs, sel, count, sink, adapter, indices); + } + + template + static idx_t SelectOutputDispatch(const std::array &inputs, const SelectionVector &sel, + idx_t count, SelectionVector *true_sel, SelectionVector *false_sel, + const InputProfile &profile, ADAPTER &adapter, + std::index_sequence indices) { + if constexpr (POLICY::SPECIALIZE_OUTPUTS) { + if (true_sel && false_sel) { + StaticSelectionSink sink(true_sel, false_sel); + return SelectInternal(inputs, sel, count, profile, sink, + adapter, indices); + } else if (true_sel) { + StaticSelectionSink sink(true_sel, false_sel); + return SelectInternal(inputs, sel, count, profile, sink, + adapter, indices); + } + StaticSelectionSink sink(true_sel, false_sel); + return SelectInternal(inputs, sel, count, profile, sink, adapter, + indices); + } + RuntimeSelectionSink sink(true_sel, false_sel); + return SelectInternal(inputs, sel, count, profile, sink, adapter, + indices); + } + + template + DUCKDB_SCALAR_EXECUTOR_NOINLINE static idx_t + SelectFallback(const std::array &inputs, const SelectionVector &sel, idx_t count, + SelectionVector *true_sel, SelectionVector *false_sel, const InputProfile &profile, ADAPTER &adapter, + std::index_sequence indices) { + return SelectOutputDispatch(inputs, sel, count, true_sel, false_sel, profile, adapter, + indices); + } + +public: + template + static void Execute(const std::array &inputs, Vector &result, idx_t count, + ADAPTER &adapter) { + ExecuteInternal(inputs, result, count, adapter, + std::index_sequence_for {}); + } + + template + static idx_t Select(const std::array &inputs, const SelectionVector *sel, idx_t count, + SelectionVector *true_sel, SelectionVector *false_sel, ADAPTER &adapter) { + if (!true_sel && !false_sel) { + throw InternalException("Either true or false sel must be set"); + } + if (!sel) { + sel = FlatVector::IncrementalSelectionVector(); + } + auto indices = std::index_sequence_for {}; + auto profile = GetInputProfile(inputs, indices); + if constexpr (POLICY::DIRECT_TRUE_FLAT_MASKS != 0) { + static_assert(sizeof...(ARGS) <= 3, "Direct true-only selection supports up to three inputs"); + if (true_sel && !false_sel) { + if (profile.all_flat_or_constant && + (POLICY::DIRECT_TRUE_FLAT_MASKS & (uint64_t(1) << profile.constant_mask))) { + StaticSelectionSink sink(true_sel, false_sel); + if (profile.any_constant_null) { + return sink.FillConstant(false, *sel, count); + } + idx_t result; + if (TrySelectFlat( + inputs, *sel, count, profile.constant_mask, sink, adapter, indices, result)) { + return result; + } + } + return SelectFallback(inputs, *sel, count, true_sel, false_sel, profile, + adapter, indices); + } + } + return SelectOutputDispatch(inputs, *sel, count, true_sel, false_sel, profile, + adapter, indices); + } +}; + +} // namespace duckdb + +#undef DUCKDB_SCALAR_EXECUTOR_NOINLINE diff --git a/src/duckdb/src/include/duckdb/common/vector_operations/unary_executor.hpp b/src/duckdb/src/include/duckdb/common/vector_operations/unary_executor.hpp index 9597eec3e..4aa6333e5 100644 --- a/src/duckdb/src/include/duckdb/common/vector_operations/unary_executor.hpp +++ b/src/duckdb/src/include/duckdb/common/vector_operations/unary_executor.hpp @@ -3,23 +3,20 @@ // // duckdb/common/vector_operations/unary_executor.hpp // -// //===----------------------------------------------------------------------===// #pragma once -#include "duckdb/common/exception.hpp" +#include "duckdb/common/enums/function_errors.hpp" #include "duckdb/common/optional.hpp" #include "duckdb/common/smaller_binary.hpp" -#include "duckdb/common/types/vector.hpp" -#include "duckdb/common/vector/constant_vector.hpp" #include "duckdb/common/vector/dictionary_vector.hpp" -#include "duckdb/common/vector/flat_vector.hpp" #include "duckdb/common/vector/string_vector.hpp" +#include "duckdb/common/vector_operations/scalar_executor.hpp" #include "duckdb/common/vector_operations/vector_operations.hpp" -#include "duckdb/common/enums/function_errors.hpp" #include +#include namespace duckdb { @@ -61,202 +58,121 @@ struct UnaryStringOperator { } }; -struct UnaryExecutor { -private: - template - static inline void ExecuteLoop(const INPUT_TYPE *__restrict ldata, RESULT_TYPE *__restrict result_data, idx_t count, - const SelectionVector *__restrict sel_vector, const ValidityMask &mask, - ValidityMask &result_mask, DATA_TYPE &data, bool adds_nulls) { -#ifdef DEBUG - // ldata may point to a compressed dictionary buffer which can be smaller than ldata + count - idx_t max_index = 0; - for (idx_t i = 0; i < count; i++) { - auto idx = sel_vector->get_index(i); - max_index = MaxValue(max_index, idx); - } - ASSERT_RESTRICT(ldata, ldata + max_index, result_data, result_data + count); -#endif +template +struct UnaryScalarAdapter { + static constexpr bool ADDS_NULLS = CAN_ADD_NULLS; - if (mask.CanHaveNull()) { - for (idx_t i = 0; i < count; i++) { - auto idx = sel_vector->get_index(i); - if (mask.RowIsValidUnsafe(idx)) { - result_data[i] = - OPWRAPPER::template Operation(ldata[idx], result_mask, i, data); - } else { - result_mask.SetInvalid(i); - } - } - } else { - for (idx_t i = 0; i < count; i++) { - auto idx = sel_vector->get_index(i); - result_data[i] = - OPWRAPPER::template Operation(ldata[idx], result_mask, i, data); - } - } + explicit UnaryScalarAdapter(DATA_TYPE &data_p) : data(data_p) { } -#if !DUCKDB_SMALLER_BINARY(unary_executor_flat) - template - static inline void ExecuteFlat(const INPUT_TYPE *__restrict ldata, RESULT_TYPE *__restrict result_data, idx_t count, - const ValidityMask &mask, ValidityMask &result_mask, DATA_TYPE &data, - bool adds_nulls) { - ASSERT_RESTRICT(ldata, ldata + count, result_data, result_data + count); + inline RESULT_TYPE Operation(ValidityMask &mask, idx_t idx, INPUT_TYPE input) { + return OPWRAPPER::template Operation(input, mask, idx, data); + } - if (mask.CanHaveNull()) { - if (!adds_nulls) { - result_mask.Initialize(mask); - } else { - result_mask.Copy(mask, count); - } - idx_t base_idx = 0; - auto entry_count = ValidityMask::EntryCount(count); - for (idx_t entry_idx = 0; entry_idx < entry_count; entry_idx++) { - auto validity_entry = mask.GetValidityEntry(entry_idx); - idx_t next = MinValue(base_idx + ValidityMask::BITS_PER_VALUE, count); - if (ValidityMask::AllValid(validity_entry)) { - // all valid: perform operation - for (; base_idx < next; base_idx++) { - result_data[base_idx] = OPWRAPPER::template Operation( - ldata[base_idx], result_mask, base_idx, data); - } - } else if (ValidityMask::NoneValid(validity_entry)) { - // nothing valid: skip all - base_idx = next; - continue; - } else { - // partially valid: need to check individual elements for validity - idx_t start = base_idx; - for (; base_idx < next; base_idx++) { - if (ValidityMask::RowIsValid(validity_entry, base_idx - start)) { - D_ASSERT(mask.RowIsValid(base_idx)); - result_data[base_idx] = OPWRAPPER::template Operation( - ldata[base_idx], result_mask, base_idx, data); - } - } - } - } - } else { - for (idx_t i = 0; i < count; i++) { - result_data[i] = - OPWRAPPER::template Operation(ldata[i], result_mask, i, data); - } - } + DATA_TYPE &data; +}; + +template +struct UnarySelectAdapter { + explicit UnarySelectAdapter(FUNC &fun_p) : fun(fun_p) { } -#endif - template - static inline void ExecuteStandard(const Vector &input, Vector &result, idx_t count, DATA_TYPE &data, - bool adds_nulls, - FunctionErrors errors = FunctionErrors::CAN_THROW_RUNTIME_ERROR) { - switch (input.GetVectorType()) { - case VectorType::CONSTANT_VECTOR: { - result.SetVectorType(VectorType::CONSTANT_VECTOR); - if (result.size() != count) { - FlatVector::SetSize(result, count); - } - auto result_data = ConstantVector::GetData(result); - auto ldata = ConstantVector::GetData(input); + inline bool Operation(INPUT_TYPE input) { + return fun(input); + } - if (ConstantVector::IsNull(input)) { - ConstantVector::SetNull(result, count_t(count)); - } else { - ConstantVector::SetNull(result, false); - *result_data = OPWRAPPER::template Operation( - *ldata, ConstantVector::Validity(result), 0, data); - } - break; - } + inline bool OperationNoNull(INPUT_TYPE input) { + return Operation(input); + } + + FUNC &fun; +}; + +struct UnaryExecutor { +private: + struct ExecutePolicy { #if !DUCKDB_SMALLER_BINARY(unary_executor_flat) - case VectorType::FLAT_VECTOR: { - result.SetVectorType(VectorType::FLAT_VECTOR); - if (result.size() != count) { - FlatVector::SetSize(result, count); - } - auto result_data = FlatVector::GetDataMutable(result); - auto ldata = FlatVector::GetData(input); + static constexpr bool SPECIALIZE_FLAT = true; +#else + static constexpr bool SPECIALIZE_FLAT = false; +#endif + static constexpr bool SPECIALIZE_NULLABLE_GENERIC_SELECTIONS = false; + static constexpr bool PRESERVE_RESULT_VALIDITY = true; + }; + + struct SelectPolicy { +#if !DUCKDB_SMALLER_BINARY(unary_executor_select_flat) + static constexpr uint64_t SPECIALIZED_MASKS = 1; +#else + static constexpr uint64_t SPECIALIZED_MASKS = 0; +#endif +#if !DUCKDB_SMALLER_BINARY(unary_executor_select_flags) + static constexpr bool SPECIALIZE_OUTPUTS = true; +#else + static constexpr bool SPECIALIZE_OUTPUTS = false; +#endif + static constexpr uint64_t DIRECT_TRUE_FLAT_MASKS = 0; + }; - ExecuteFlat(ldata, result_data, count, FlatVector::Validity(input), - FlatVector::ValidityMutable(result), data, adds_nulls); - break; - } - case VectorType::DICTIONARY_VECTOR: { - // dictionary vector - we can run the function ONLY on the dictionary in some cases - // we can only do this if the function does not throw errors - // we can execute the function on a value that is in the dictionary but that is not referenced - // if the function can throw errors - this will result in us (incorrectly) throwing an error - if (errors == FunctionErrors::CANNOT_ERROR) { - static constexpr idx_t DICTIONARY_THRESHOLD = 2; - auto dict_size = DictionaryVector::DictionarySize(input); - if (dict_size.IsValid() && dict_size.GetIndex() * DICTIONARY_THRESHOLD <= count) { - // we can operate directly on the dictionary if we have a dictionary size - // but this only makes sense if the dictionary size is smaller than the count by some factor - auto &dictionary_values = DictionaryVector::Child(input); - if (dictionary_values.GetVectorType() == VectorType::FLAT_VECTOR) { - // execute the function over the dictionary - auto result_data = FlatVector::GetDataMutable(result); - auto ldata = FlatVector::GetData(dictionary_values); - ExecuteFlat( - ldata, result_data, dict_size.GetIndex(), FlatVector::Validity(dictionary_values), - FlatVector::ValidityMutable(result), data, adds_nulls); - // slice the result with the original offsets - auto &offsets = DictionaryVector::SelVector(input); - FlatVector::SetSize(result, dict_size.GetIndex()); - result.Dictionary(result, dict_size.GetIndex(), offsets, count); - break; - } + template + static inline void ExecuteInternal(const Vector &input, Vector &result, idx_t count, DATA_TYPE &data, + FunctionErrors errors) { + UnaryScalarAdapter adapter(data); + +#if !DUCKDB_SMALLER_BINARY(unary_executor_flat) + if (input.GetVectorType() == VectorType::DICTIONARY_VECTOR && errors == FunctionErrors::CANNOT_ERROR) { + static constexpr idx_t DICTIONARY_THRESHOLD = 2; + auto dictionary_size = DictionaryVector::DictionarySize(input); + if (dictionary_size.IsValid() && dictionary_size.GetIndex() * DICTIONARY_THRESHOLD <= count) { + auto &dictionary_values = DictionaryVector::Child(input); + if (dictionary_values.GetVectorType() == VectorType::FLAT_VECTOR) { + std::array dictionary_input = {{dictionary_values}}; + ScalarExecutor::Execute( + dictionary_input, result, dictionary_size.GetIndex(), adapter); + auto &offsets = DictionaryVector::SelVector(input); + FlatVector::SetSize(result, dictionary_size.GetIndex()); + result.Dictionary(result, dictionary_size.GetIndex(), offsets, count); + return; } } - DUCKDB_EXPLICIT_FALLTHROUGH; } #endif - default: { - UnifiedVectorFormat vdata; - input.ToUnifiedFormat(vdata); - result.SetVectorType(VectorType::FLAT_VECTOR); - if (result.size() != count) { - FlatVector::SetSize(result, count); - } - auto result_data = FlatVector::GetDataMutable(result); - auto ldata = UnifiedVectorFormat::GetData(vdata); - - ExecuteLoop(ldata, result_data, count, vdata.sel, vdata.validity, - FlatVector::ValidityMutable(result), data, adds_nulls); - break; - } - } + std::array inputs = {{input}}; + ScalarExecutor::Execute(inputs, result, count, + adapter); } public: template static void Execute(const Vector &input, Vector &result, idx_t count) { std::nullptr_t no_data = nullptr; - ExecuteStandard(input, result, count, no_data, false); + ExecuteInternal( + input, result, count, no_data, FunctionErrors::CAN_THROW_RUNTIME_ERROR); } template > static void Execute(const Vector &input, Vector &result, idx_t count, FUNC fun, FunctionErrors errors = FunctionErrors::CAN_THROW_RUNTIME_ERROR) { constexpr bool adds_nulls = - std::is_same())), optional>::value; - ExecuteStandard(input, result, count, fun, adds_nulls, - errors); + std::is_same, optional>::value; + ExecuteInternal(input, result, count, fun, + errors); } template - static void GenericExecute(const Vector &input, Vector &result, idx_t count, DATA_TYPE &data, - bool adds_nulls = false) { - ExecuteStandard(input, result, count, data, adds_nulls); + static void GenericExecute(const Vector &input, Vector &result, idx_t count, DATA_TYPE &data, bool = false) { + // Generic operations own the result mask so they can invalidate rows at runtime. + ExecuteInternal( + input, result, count, data, FunctionErrors::CAN_THROW_RUNTIME_ERROR); } template static void ExecuteString(const Vector &input, Vector &result, idx_t count) { auto &heap = StringVector::GetStringHeap(result); - UnaryExecutor::GenericExecute>(input, result, count, heap); + GenericExecute>(input, result, count, heap); } - //! Convenience overloads without explicit count - count is derived from input.size(). template static void Execute(const Vector &input, Vector &result) { Execute(input, result, input.size()); @@ -278,76 +194,13 @@ struct UnaryExecutor { ExecuteString(input, result, input.size()); } -private: - // Select logic copied from TernaryExecutor, but with a lambda instead of a static functor - template , bool NO_NULL, bool HAS_TRUE_SEL, - bool HAS_FALSE_SEL> - static inline idx_t SelectLoop(const INPUT_TYPE *__restrict input_data, const SelectionVector *result_sel, - const idx_t count, FUNC fun, const SelectionVector &input_sel, - const ValidityMask &input_validity, SelectionVector *true_sel, - SelectionVector *false_sel) { - idx_t true_count = 0, false_count = 0; - for (idx_t i = 0; i < count; i++) { - const auto result_idx = result_sel->get_index(i); - const auto idx = input_sel.get_index(i); - const bool comparison_result = (NO_NULL || input_validity.RowIsValid(idx)) && fun(input_data[idx]); - if (HAS_TRUE_SEL) { - true_sel->set_index(true_count, result_idx); - true_count += comparison_result; - } - if (HAS_FALSE_SEL) { - false_sel->set_index(false_count, result_idx); - false_count += !comparison_result; - } - } - if (HAS_TRUE_SEL) { - return true_count; - } else { - return count - false_count; - } - } - - template , bool NO_NULL> - static inline idx_t SelectLoopSelSwitch(UnifiedVectorFormat &input_data, const SelectionVector *sel, - const idx_t count, FUNC fun, SelectionVector *true_sel, - SelectionVector *false_sel) { - if (true_sel && false_sel) { - return SelectLoop( - UnifiedVectorFormat::GetData(input_data), sel, count, fun, *input_data.sel, - input_data.validity, true_sel, false_sel); - } else if (true_sel) { - return SelectLoop( - UnifiedVectorFormat::GetData(input_data), sel, count, fun, *input_data.sel, - input_data.validity, true_sel, false_sel); - } else { - D_ASSERT(false_sel); - return SelectLoop( - UnifiedVectorFormat::GetData(input_data), sel, count, fun, *input_data.sel, - input_data.validity, true_sel, false_sel); - } - } - template > - static inline idx_t SelectLoopSwitch(UnifiedVectorFormat &input_data, const SelectionVector *sel, const idx_t count, - FUNC fun, SelectionVector *true_sel, SelectionVector *false_sel) { - if (input_data.validity.CanHaveNull()) { - return SelectLoopSelSwitch(input_data, sel, count, fun, true_sel, false_sel); - } else { - return SelectLoopSelSwitch(input_data, sel, count, fun, true_sel, false_sel); - } - } - -public: - template > - static idx_t Select(const Vector &input, const SelectionVector *sel, const idx_t count, FUNC fun, + static idx_t Select(const Vector &input, const SelectionVector *sel, idx_t count, FUNC fun, SelectionVector *true_sel, SelectionVector *false_sel) { - if (!sel) { - sel = FlatVector::IncrementalSelectionVector(); - } - UnifiedVectorFormat input_data; - input.ToUnifiedFormat(input_data); - - return SelectLoopSwitch(input_data, sel, count, fun, true_sel, false_sel); + std::array inputs = {{input}}; + UnarySelectAdapter adapter(fun); + return ScalarExecutor::Select(inputs, sel, count, true_sel, + false_sel, adapter); } }; diff --git a/src/duckdb/src/include/duckdb/common/vector_operations/variadic_executor.hpp b/src/duckdb/src/include/duckdb/common/vector_operations/variadic_executor.hpp index 7fb9d0e04..190641122 100644 --- a/src/duckdb/src/include/duckdb/common/vector_operations/variadic_executor.hpp +++ b/src/duckdb/src/include/duckdb/common/vector_operations/variadic_executor.hpp @@ -3,26 +3,24 @@ // // duckdb/common/vector_operations/variadic_executor.hpp // -// //===----------------------------------------------------------------------===// #pragma once #include "duckdb/common/optional.hpp" +#include "duckdb/common/smaller_binary.hpp" #include "duckdb/common/types/data_chunk.hpp" -#include "duckdb/common/types/vector.hpp" -#include "duckdb/common/vector/constant_vector.hpp" -#include "duckdb/common/vector/flat_vector.hpp" +#include "duckdb/common/vector_operations/scalar_executor.hpp" #include #include #include +#include namespace duckdb { //! Wrappers that adapt different calling conventions to a uniform interface. //! Each wrapper's Operation method takes: (FUN, ValidityMask&, idx_t, ARGS...) - struct VariadicLambdaWrapper { template static inline RESULT_TYPE Operation(FUN &fun, ValidityMask &mask, idx_t idx, ARGS... args) { @@ -47,179 +45,96 @@ struct VariadicStandardOperatorWrapper { } }; -//! VariadicExecutor: a unified executor for any number of input vectors. -//! Uses C++17 fold expressions and std::index_sequence to generalize the -//! pattern shared by TernaryExecutor, SenaryExecutor, and SeptenaryExecutor. -//! -//! Template parameter ordering: -//! This differs from TernaryExecutor's because -//! a parameter pack must be last (or followed only by deducible params). -//! The named executors (TernaryExecutor, etc.) provide backward-compatible APIs. -struct VariadicExecutor { - using VectorRef = std::reference_wrapper; - -private: - template - static std::array MakeInputArrayImpl(DataChunk &input, std::index_sequence) { - return {{std::cref(input.data[Is])...}}; - } - - template - static std::array MakeInputArray(DataChunk &input) { - D_ASSERT(input.ColumnCount() >= N); - return MakeInputArrayImpl(input, std::make_index_sequence {}); - } +template +struct VariadicLambdaAdapter { + static constexpr bool ADDS_NULLS = + std::is_same, optional>::value; - template - static bool AllConstant(const std::array &inputs) { - for (size_t i = 0; i < N; i++) { - if (inputs[i].get().GetVectorType() != VectorType::CONSTANT_VECTOR) { - return false; - } - } - return true; + explicit VariadicLambdaAdapter(FUN &fun_p) : fun(fun_p) { } - template - static bool AnyConstantNull(const std::array &inputs) { - for (size_t i = 0; i < N; i++) { - if (ConstantVector::IsNull(inputs[i].get())) { - return true; - } - } - return false; + inline RESULT_TYPE Operation(ValidityMask &mask, idx_t idx, ARGS... args) { + return VariadicLambdaWrapper::template Operation(fun, mask, idx, args...); } - //------------------------------------------------------------------- - // Execute implementation - //------------------------------------------------------------------- - template - static void ExecuteImplWithIndices(std::array &inputs, Vector &result, idx_t count, - FUN fun, std::index_sequence) { - constexpr size_t N = sizeof...(ARGS); - - if (AllConstant(inputs)) { - result.SetVectorType(VectorType::CONSTANT_VECTOR); - FlatVector::SetSize(result, count); - if (AnyConstantNull(inputs)) { - ConstantVector::SetNull(result, true); - } else { - auto result_data = ConstantVector::GetData(result); - result_data[0] = OPWRAPPER::template Operation( - fun, ConstantVector::Validity(result), 0, *ConstantVector::GetData(inputs[Is].get())...); - } - } else { - result.SetVectorType(VectorType::FLAT_VECTOR); - auto result_data = FlatVector::GetDataMutable(result); - auto &result_validity = FlatVector::ValidityMutable(result); - - std::array vdata; - for (size_t i = 0; i < N; i++) { - inputs[i].get().ToUnifiedFormat(vdata[i]); - } + FUN &fun; +}; - auto data_ptrs = std::make_tuple(UnifiedVectorFormat::GetData(vdata[Is])...); +template +struct VariadicStandardAdapter { + static constexpr bool ADDS_NULLS = false; - if ((... || vdata[Is].validity.CanHaveNull())) { - for (idx_t i = 0; i < count; i++) { - std::array idxs = {{vdata[Is].sel->get_index(i)...}}; - if ((... && vdata[Is].validity.RowIsValid(idxs[Is]))) { - result_data[i] = OPWRAPPER::template Operation( - fun, result_validity, i, std::get(data_ptrs)[idxs[Is]]...); - } else { - result_validity.SetInvalid(i); - } - } - } else { - for (idx_t i = 0; i < count; i++) { - result_data[i] = OPWRAPPER::template Operation( - fun, result_validity, i, std::get(data_ptrs)[vdata[Is].sel->get_index(i)]...); - } - } - FlatVector::SetSize(result, count); - } + inline RESULT_TYPE Operation(ValidityMask &, idx_t, ARGS... args) { + return OP::template Operation(args...); } +}; - //------------------------------------------------------------------- - // Select implementation - //------------------------------------------------------------------- - template - static idx_t SelectLoopImpl(std::tuple &data_ptrs, - std::array &vdata, - const SelectionVector *result_sel, idx_t count, SelectionVector *true_sel, - SelectionVector *false_sel, std::index_sequence) { - constexpr size_t N = sizeof...(ARGS); - idx_t true_count = 0, false_count = 0; - for (idx_t i = 0; i < count; i++) { - auto result_idx = result_sel->get_index(i); - std::array idxs = {{vdata[Is].sel->get_index(i)...}}; - bool comparison_result = (NO_NULL || (... && vdata[Is].validity.RowIsValid(idxs[Is]))) && - OP::Operation(std::get(data_ptrs)[idxs[Is]]...); - if (HAS_TRUE_SEL) { - true_sel->set_index(true_count, result_idx); - true_count += comparison_result; - } - if (HAS_FALSE_SEL) { - false_sel->set_index(false_count, result_idx); - false_count += !comparison_result; - } - } - if (HAS_TRUE_SEL) { - return true_count; - } - return count - false_count; +template +struct VariadicSelectAdapter { + inline bool Operation(ARGS... args) { + return OP::Operation(args...); } - template - static idx_t SelectLoopSelSwitch(std::tuple &data_ptrs, - std::array &vdata, - const SelectionVector *sel, idx_t count, SelectionVector *true_sel, - SelectionVector *false_sel, std::index_sequence indices) { - if (true_sel && false_sel) { - return SelectLoopImpl(data_ptrs, vdata, sel, count, true_sel, false_sel, - indices); - } else if (true_sel) { - return SelectLoopImpl(data_ptrs, vdata, sel, count, true_sel, false_sel, - indices); - } else { - D_ASSERT(false_sel); - return SelectLoopImpl(data_ptrs, vdata, sel, count, true_sel, false_sel, - indices); - } + inline bool OperationNoNull(ARGS... args) { + return Operation(args...); } +}; - template - static idx_t SelectImplWithIndices(std::array &inputs, const SelectionVector *sel, - idx_t count, SelectionVector *true_sel, SelectionVector *false_sel, - std::index_sequence indices) { - constexpr size_t N = sizeof...(ARGS); - if (!sel) { - sel = FlatVector::IncrementalSelectionVector(); - } +//! VariadicExecutor is the generic public facade over ScalarExecutor. +//! Template parameter ordering remains . +struct VariadicExecutor { + using VectorRef = ScalarExecutor::VectorRef; - std::array vdata; - for (size_t i = 0; i < N; i++) { - inputs[i].get().ToUnifiedFormat(vdata[i]); - } +private: + template + static constexpr bool SpecializeFlat() { + return sizeof...(ARGS) <= 3 && (... && std::is_arithmetic::value); + } - auto data_ptrs = std::make_tuple(UnifiedVectorFormat::GetData(vdata[Is])...); + template + struct ExecutePolicy { +#if !DUCKDB_SMALLER_BINARY(variadic_executor_flat) + static constexpr bool SPECIALIZE_FLAT = SpecializeFlat(); +#else + static constexpr bool SPECIALIZE_FLAT = false; +#endif + static constexpr bool SPECIALIZE_NULLABLE_GENERIC_SELECTIONS = false; + static constexpr bool PRESERVE_RESULT_VALIDITY = false; + }; + + template + struct SelectPolicy { +#if !DUCKDB_SMALLER_BINARY(variadic_executor_select_flat) + static constexpr uint64_t SPECIALIZED_MASKS = SpecializeFlat() ? 1 : 0; + static constexpr uint64_t DIRECT_TRUE_FLAT_MASKS = + sizeof...(ARGS) == 3 && SpecializeFlat() ? uint64_t(1) << 6 : 0; +#else + static constexpr uint64_t SPECIALIZED_MASKS = 0; + static constexpr uint64_t DIRECT_TRUE_FLAT_MASKS = 0; +#endif +#if !DUCKDB_SMALLER_BINARY(variadic_executor_select_flags) + static constexpr bool SPECIALIZE_OUTPUTS = true; +#else + static constexpr bool SPECIALIZE_OUTPUTS = false; +#endif + }; - if ((... || vdata[Is].validity.CanHaveNull())) { - return SelectLoopSelSwitch(data_ptrs, vdata, sel, count, true_sel, false_sel, indices); - } else { - return SelectLoopSelSwitch(data_ptrs, vdata, sel, count, true_sel, false_sel, indices); - } + template + static std::array MakeInputArrayImpl(DataChunk &input, std::index_sequence) { + return {{std::cref(input.data[Is])...}}; + } + + template + static std::array MakeInputArray(DataChunk &input) { + D_ASSERT(input.ColumnCount() >= N); + return MakeInputArrayImpl(input, std::make_index_sequence {}); } -private: - //------------------------------------------------------------------- - // Verify all inputs have the same size and return that size. - //------------------------------------------------------------------- template static idx_t CheckExecuteCount(const std::array &inputs) { static_assert(N > 0, "VariadicExecutor requires at least one input"); idx_t count = inputs[0].get().size(); - for (size_t i = 1; i < N; i++) { + for (idx_t i = 1; i < N; i++) { if (inputs[i].get().size() != count) { throw InternalException( "Mismatch in input vector sizes for VariadicExecutor - expected %d rows but got %d", count, @@ -230,47 +145,36 @@ struct VariadicExecutor { } public: - //------------------------------------------------------------------- - // Execute: lambda-based, with Vector array - //------------------------------------------------------------------- template static void Execute(std::array inputs, Vector &result, FUN fun) { - constexpr size_t N = sizeof...(ARGS); - const idx_t count = CheckExecuteCount(inputs); - ExecuteImplWithIndices(inputs, result, count, fun, - std::index_sequence_for {}); + auto count = CheckExecuteCount(inputs); + VariadicLambdaAdapter adapter(fun); + ScalarExecutor::Execute, RESULT_TYPE, decltype(adapter), ARGS...>(inputs, result, count, + adapter); } - //------------------------------------------------------------------- - // Execute: lambda-based, with DataChunk - //------------------------------------------------------------------- template static void Execute(DataChunk &input, Vector &result, FUN fun) { auto inputs = MakeInputArray(input); - ExecuteImplWithIndices(inputs, result, input.size(), fun, - std::index_sequence_for {}); + VariadicLambdaAdapter adapter(fun); + ScalarExecutor::Execute, RESULT_TYPE, decltype(adapter), ARGS...>(inputs, result, + input.size(), adapter); } - //------------------------------------------------------------------- - // ExecuteStandard: static OP::Operation, with Vector array - //------------------------------------------------------------------- template static void ExecuteStandard(std::array inputs, Vector &result) { - constexpr size_t N = sizeof...(ARGS); - const idx_t count = CheckExecuteCount(inputs); - bool dummy = false; - ExecuteImplWithIndices, bool, ARGS...>( - inputs, result, count, dummy, std::index_sequence_for {}); + auto count = CheckExecuteCount(inputs); + VariadicStandardAdapter adapter; + ScalarExecutor::Execute, RESULT_TYPE, decltype(adapter), ARGS...>(inputs, result, count, + adapter); } - //------------------------------------------------------------------- - // Select: OP::Operation returns bool, with Vector array - //------------------------------------------------------------------- template static idx_t Select(std::array inputs, const SelectionVector *sel, idx_t count, SelectionVector *true_sel, SelectionVector *false_sel) { - return SelectImplWithIndices(inputs, sel, count, true_sel, false_sel, - std::index_sequence_for {}); + VariadicSelectAdapter adapter; + return ScalarExecutor::Select, decltype(adapter), ARGS...>(inputs, sel, count, true_sel, + false_sel, adapter); } }; diff --git a/src/duckdb/src/include/duckdb/execution/operator/join/physical_hash_join.hpp b/src/duckdb/src/include/duckdb/execution/operator/join/physical_hash_join.hpp index 445624c93..733ca89e6 100644 --- a/src/duckdb/src/include/duckdb/execution/operator/join/physical_hash_join.hpp +++ b/src/duckdb/src/include/duckdb/execution/operator/join/physical_hash_join.hpp @@ -81,6 +81,12 @@ class PhysicalHashJoin : public PhysicalComparisonJoin { bool ParallelOperator() const override { return true; } + PipelineExternalInputSupport GetExternalInputSupport() const override { + return PipelineExternalInputSupport::SUPPORTED; + } + PipelineExternalInputCost GetExternalInputCost() const override { + return PipelineExternalInputCost::SERIALIZED_FANOUT; + } protected: // CachingOperator Interface diff --git a/src/duckdb/src/include/duckdb/execution/operator/set/physical_cte.hpp b/src/duckdb/src/include/duckdb/execution/operator/set/physical_cte.hpp index c945ecc73..169fde9f3 100644 --- a/src/duckdb/src/include/duckdb/execution/operator/set/physical_cte.hpp +++ b/src/duckdb/src/include/duckdb/execution/operator/set/physical_cte.hpp @@ -118,7 +118,11 @@ class PhysicalCTE : public PhysicalOperator { public: void BuildPipelines(Pipeline ¤t, MetaPipeline &meta_pipeline) override; + bool CanRegisterDirectConsumer(Pipeline &pipeline) const; + void RegisterDirectConsumer(Pipeline &pipeline, idx_t consumer_idx); bool TryRegisterDirectConsumer(Pipeline &pipeline, idx_t consumer_idx); + vector> GetProducerPipelines() const; + void SetPipelineSelectionResolved(); bool ShouldUseBufferedConsumer(Pipeline &pipeline) const; void RegisterBufferedConsumer(Pipeline &pipeline, idx_t consumer_idx); void RegisterMaterializedConsumer(idx_t consumer_idx); diff --git a/src/duckdb/src/include/duckdb/execution/physical_operator.hpp b/src/duckdb/src/include/duckdb/execution/physical_operator.hpp index 602024491..3e9575656 100644 --- a/src/duckdb/src/include/duckdb/execution/physical_operator.hpp +++ b/src/duckdb/src/include/duckdb/execution/physical_operator.hpp @@ -37,6 +37,7 @@ class PhysicalPlan; enum class TableFunctionParallelism : uint8_t; enum class OperatorCachingMode : uint8_t { NONE, PARTITIONED, ORDERED, UNORDERED }; enum class PipelineExternalInputSupport : uint8_t { UNSUPPORTED, SUPPORTED }; +enum class PipelineExternalInputCost : uint8_t { PIPELINED, SERIALIZED_FANOUT }; enum class PipelineSourceConsumption : uint8_t { ALL_INPUT, MAY_STOP_EARLY }; //! PhysicalOperator is the base class of the physical operators present in the execution plan. @@ -115,6 +116,9 @@ class PhysicalOperator { virtual PipelineExternalInputSupport GetExternalInputSupport() const { return PipelineExternalInputSupport::UNSUPPORTED; } + virtual PipelineExternalInputCost GetExternalInputCost() const { + return PipelineExternalInputCost::PIPELINED; + } virtual PipelineSourceConsumption GetSourceConsumption() const { return PipelineSourceConsumption::ALL_INPUT; diff --git a/src/duckdb/src/include/duckdb/parallel/meta_pipeline.hpp b/src/duckdb/src/include/duckdb/parallel/meta_pipeline.hpp index f74dac103..a68454065 100644 --- a/src/duckdb/src/include/duckdb/parallel/meta_pipeline.hpp +++ b/src/duckdb/src/include/duckdb/parallel/meta_pipeline.hpp @@ -21,6 +21,18 @@ enum class MetaPipelineType : uint8_t { enum class MetaPipelineDependencyMode : uint8_t { ADD_DEPENDENCY, NO_DEPENDENCY }; enum class RecursiveDependencyMode : uint8_t { RESPECT_PARALLELISM, FORCE }; enum class DataflowDependencyMode : uint8_t { INCLUDE, SKIP_CONFLICTING }; +enum class MetaPipelineDependencyType : uint8_t { REQUIRED, OPTIONAL_DEPENDENCY }; + +struct MetaPipelineDependency { + MetaPipelineDependency(Pipeline &pipeline_p, MetaPipelineDependencyType type_p) + : pipeline(pipeline_p), type(type_p) { + } + + reference pipeline; + MetaPipelineDependencyType type; +}; + +using meta_pipeline_dependency_map_t = reference_map_t>; //! MetaPipeline represents a set of pipelines that all have the same sink class MetaPipeline : public enable_shared_from_this { @@ -57,7 +69,9 @@ class MetaPipeline : public enable_shared_from_this { //! Recursively gets the last child added MetaPipeline &GetLastChild(); //! Get the dependencies of the Pipelines of this MetaPipeline - const reference_map_t>> &GetDependencies() const; + const meta_pipeline_dependency_map_t &GetDependencies() const; + bool RemoveOptionalDependency(Pipeline &pipeline, Pipeline &dependency); + void AddOptionalDependency(Pipeline &pipeline, Pipeline &dependency); //! Whether the sink of this pipeline is a join build MetaPipelineType Type() const; //! Whether this MetaPipeline has a recursive CTE @@ -68,7 +82,9 @@ class MetaPipeline : public enable_shared_from_this { void AssignNextBatchIndex(Pipeline &pipeline); //! Let 'dependant' depend on all pipeline that were created since 'start', //! where 'including' determines whether 'start' is added to the dependencies - vector> AddDependenciesFrom(Pipeline &dependant, const Pipeline &start, bool including); + vector> + AddDependenciesFrom(Pipeline &dependant, const Pipeline &start, bool including, + MetaPipelineDependencyType dependency_type = MetaPipelineDependencyType::REQUIRED); //! Recursively makes all children of this MetaPipeline depend on the given Pipeline. //! Force dependencies when ordering is mandatory, rather than using the pipeline/thread-count heuristic. void @@ -118,7 +134,7 @@ class MetaPipeline : public enable_shared_from_this { //! All pipelines with a different source, but the same sink vector> pipelines; //! Dependencies of Pipelines of this MetaPipeline - reference_map_t>> pipeline_dependencies; + meta_pipeline_dependency_map_t pipeline_dependencies; //! Other MetaPipelines that this MetaPipeline depends on vector> children; //! Next batch index diff --git a/src/duckdb/src/include/duckdb/parallel/pipeline.hpp b/src/duckdb/src/include/duckdb/parallel/pipeline.hpp index 3840c6727..7ba9d9817 100644 --- a/src/duckdb/src/include/duckdb/parallel/pipeline.hpp +++ b/src/duckdb/src/include/duckdb/parallel/pipeline.hpp @@ -24,6 +24,8 @@ class Event; class MetaPipeline; class PipelineExecutor; class Pipeline; +class PipelineBuildStateData; +class PhysicalCTE; enum class PipelineInputMode : uint8_t { SCHEDULED_SOURCE, EXTERNAL_INPUT }; enum class ExternalInputEventState : uint8_t { @@ -61,6 +63,10 @@ class PipelineBuildState { //! How much to increment batch indexes when multiple pipelines share the same source constexpr static idx_t BATCH_INCREMENT = 10000000000000; +public: + PipelineBuildState(); + ~PipelineBuildState(); // NOLINT: PipelineBuildStateData is incomplete here + public: //! Duplicate eliminated join scan dependencies reference_map_t> delim_join_dependencies; @@ -77,6 +83,15 @@ class PipelineBuildState { optional_ptr GetPipelineSource(Pipeline &pipeline); optional_ptr GetPipelineSink(Pipeline &pipeline); vector> GetPipelineOperators(Pipeline &pipeline); + void AddExternalInputCandidate(Pipeline &pipeline, PhysicalOperator &materialized_source, + PhysicalOperator &external_source, shared_ptr cte_dependency, + PhysicalCTE &cte, idx_t consumer_idx); + void AddCTEPipelineSelection(PhysicalCTE &cte, Pipeline &pipeline, shared_ptr cte_dependency, + bool dependency_added); + void ResolveExternalInputs(const vector> &meta_pipelines); + +private: + unique_ptr data; }; //! The Pipeline class represents an execution pipeline starting at a @@ -98,14 +113,10 @@ class Pipeline : public enable_shared_from_this { void AddDependency(shared_ptr &pipeline); void AddDataflowDependency(shared_ptr &pipeline); - void AddExternalFinishDependency(shared_ptr &pipeline); vector> GetDependencies() const; const vector> &GetDataflowDependencies() const { return dataflow_dependencies; } - const vector> &GetExternalFinishDependencies() const { - return external_finish_dependencies; - } bool HasDataflowDependencies() const { return !dataflow_dependencies.empty(); } @@ -148,10 +159,14 @@ class Pipeline : public enable_shared_from_this { //! Returns whether any of the operators in the pipeline care about preserving order bool IsOrderDependent() const; //! Marks this pipeline as fed externally instead of by scheduled source tasks - void SetExternalInput(); + void SetExternalInput(const vector> &producer_pipelines); bool IsExternalInput() const { return input_mode == PipelineInputMode::EXTERNAL_INPUT; } + const vector> &GetExternalInputProducers() const { + return external_input_producers; + } + bool HasExternalInputProducer(const Pipeline &pipeline) const; void SetExternalStreamingResultProducer() { external_streaming_result_producer = true; } @@ -159,6 +174,7 @@ class Pipeline : public enable_shared_from_this { void SetExternalInputEvent(const shared_ptr &event); void CompleteExternalInput(); bool CanUseExternalInput(const OperatorPartitionInfo &source_partition_info) const; + PipelineExternalInputCost GetExternalInputCost() const; bool CanStopSourceEarly() const; idx_t GetBaseBatchIndex() const { @@ -194,8 +210,8 @@ class Pipeline : public enable_shared_from_this { vector> dependencies; //! Pipelines that must be initialized before this pipeline can consume their dataflow output vector> dataflow_dependencies; - //! Pipelines that must run before this externally fed pipeline can finish its sink - vector> external_finish_dependencies; + //! Pipelines that push input into this pipeline instead of scanning its source + vector> external_input_producers; //! The base batch index of this pipeline idx_t base_batch_index = 0; @@ -218,6 +234,8 @@ class Pipeline : public enable_shared_from_this { multiset batch_indexes; private: + void RemoveDependency(const shared_ptr &pipeline); + void ClearExternalInput(); void ScheduleSequentialTask(shared_ptr &event); bool LaunchScanTasks(shared_ptr &event, idx_t max_threads); void ResetSinkAndOperators(); diff --git a/src/duckdb/src/include/duckdb/parallel/pipeline_broadcast_exchange.hpp b/src/duckdb/src/include/duckdb/parallel/pipeline_broadcast_exchange.hpp index 4b0b75195..9217af6ba 100644 --- a/src/duckdb/src/include/duckdb/parallel/pipeline_broadcast_exchange.hpp +++ b/src/duckdb/src/include/duckdb/parallel/pipeline_broadcast_exchange.hpp @@ -69,6 +69,7 @@ class PipelineBroadcastExchangeLocalState { optional_idx GetSourceMinBatchIndex(const SourcePartitionInfo &partition_info) const; vector> direct_executors; + vector> direct_input_chunks; idx_t direct_idx = 0; idx_t direct_next_batch_idx = 0; idx_t direct_min_batch_idx = 0; @@ -112,7 +113,10 @@ class PipelineBroadcastExchange { void SetProducerPipelines(const vector> &pipelines); idx_t RegisterConsumer(); + bool CanRegisterDirectConsumer(Pipeline &pipeline) const; + void SelectDirectConsumer(Pipeline &pipeline, idx_t consumer_idx); bool TryRegisterDirectConsumer(Pipeline &pipeline, idx_t consumer_idx); + vector> GetProducerPipelines() const; void SelectBufferedConsumer(idx_t consumer_idx, PipelineBroadcastExchangeScanMode scan_mode); void SelectMaterializedConsumer(idx_t consumer_idx); void ResetConsumerRegistrations(); diff --git a/src/duckdb/src/include/duckdb/parallel/pipeline_schedule.hpp b/src/duckdb/src/include/duckdb/parallel/pipeline_schedule.hpp index 0b2d72bcd..9099b07cc 100644 --- a/src/duckdb/src/include/duckdb/parallel/pipeline_schedule.hpp +++ b/src/duckdb/src/include/duckdb/parallel/pipeline_schedule.hpp @@ -9,6 +9,7 @@ #pragma once #include "duckdb/common/common.hpp" +#include "duckdb/common/optional_ptr.hpp" #include "duckdb/common/reference_map.hpp" namespace duckdb { @@ -27,9 +28,33 @@ struct PipelineScheduleStage { vector dependencies; }; +struct PipelineScheduleExternalInputDependency { + PipelineScheduleExternalInputDependency(idx_t dependent_p, idx_t dependency_index_p, Pipeline &consumer_p) + : dependent(dependent_p), dependency_index(dependency_index_p), consumer(consumer_p) { + } + + idx_t dependent; + idx_t dependency_index; + reference consumer; +}; + +struct PipelineScheduleEdge { + PipelineScheduleEdge(idx_t dependent_p, idx_t dependency_p, optional_ptr external_input_consumer_p) + : dependent(dependent_p), dependency(dependency_p), external_input_consumer(external_input_consumer_p) { + } + + idx_t dependent; + idx_t dependency; + optional_ptr external_input_consumer; +}; + struct PipelineSchedule { vector stages; vector> initialize_on_schedule_pipelines; + vector external_input_dependencies; + + bool HasCycle() const; + vector GetCycle() const; }; unique_ptr BuildPipelineSchedule(const vector> &meta_pipelines, diff --git a/src/duckdb/src/optimizer/rule/timestamp_comparison.cpp b/src/duckdb/src/optimizer/rule/timestamp_comparison.cpp index 6d3d27d35..49714aad0 100644 --- a/src/duckdb/src/optimizer/rule/timestamp_comparison.cpp +++ b/src/duckdb/src/optimizer/rule/timestamp_comparison.cpp @@ -53,6 +53,15 @@ static optional_ptr GetTimestampCast(Expression &expr) return &cast_expr; } +static Value DateToTimestampValue(date_t date, dtime_t time) { + if (date == date_t::infinity()) { + return Value::TIMESTAMP(timestamp_t::infinity()); + } else if (date == date_t::ninfinity()) { + return Value::TIMESTAMP(timestamp_t::ninfinity()); + } + return Value::TIMESTAMP(date, time); +} + unique_ptr TimeStampComparison::Apply(LogicalOperator &op, vector> &bindings, bool &changes_made, bool is_root) { auto &comparison = bindings[0].get().Cast(); @@ -82,9 +91,17 @@ unique_ptr TimeStampComparison::Apply(LogicalOperator &op, vector(original_val_ts); + if (!original_val.IsFinite()) { + auto column_copy = cast_columnref->Copy(); + auto eq_expr = BoundComparisonExpression::Create(ExpressionType::COMPARE_EQUAL, std::move(column_copy), + std::move(original_val_for_comparison)); + new_expr->GetChildrenMutable().push_back(std::move(eq_expr)); + return std::move(new_expr); + } + // add one day and validate the new date // code is inspired by AddOperator::Operation(date_t left, int32_t right). The function wasn't used directly // since it throws errors that I cannot catch here. diff --git a/src/duckdb/src/parallel/executor.cpp b/src/duckdb/src/parallel/executor.cpp index 0ac5872d7..19d1fdd0f 100644 --- a/src/duckdb/src/parallel/executor.cpp +++ b/src/duckdb/src/parallel/executor.cpp @@ -83,6 +83,9 @@ void Executor::ScheduleEventsInternal(ScheduleEventData &event_data) { D_ASSERT(events.empty()); auto schedule = BuildPipelineSchedule(event_data.meta_pipelines); + if (schedule->HasCycle()) { + throw InternalException("Cyclic dependency in pipeline schedule"); + } events.reserve(schedule->stages.size()); for (auto &stage : schedule->stages) { events.push_back(CreatePipelineScheduleEvent(stage, event_data.initial_schedule)); @@ -241,6 +244,12 @@ void Executor::InitializeInternal(PhysicalOperator &plan) { PipelineBuildState state; auto root_pipeline = make_shared_ptr(*this, state, nullptr); root_pipeline->Build(*physical_plan); + + // Resolve graph-dependent input modes after every pipeline and dependency has been constructed. + vector> to_schedule; + root_pipeline->GetMetaPipelines(to_schedule, true, true); + state.ResolveExternalInputs(to_schedule); + profiler->Initialize(plan); root_pipeline->Ready(); @@ -254,10 +263,6 @@ void Executor::InitializeInternal(PhysicalOperator &plan) { root_pipeline->GetPipelines(root_pipelines, false); root_pipeline_idx = 0; - // collect all meta-pipelines from the root pipeline - vector> to_schedule; - root_pipeline->GetMetaPipelines(to_schedule, true, true); - // number of 'PipelineCompleteEvent's is equal to the number of meta pipelines, so we have to set it here total_pipelines = to_schedule.size(); diff --git a/src/duckdb/src/parallel/meta_pipeline.cpp b/src/duckdb/src/parallel/meta_pipeline.cpp index ab4fade1c..72330aca1 100644 --- a/src/duckdb/src/parallel/meta_pipeline.cpp +++ b/src/duckdb/src/parallel/meta_pipeline.cpp @@ -62,10 +62,29 @@ MetaPipeline &MetaPipeline::GetLastChild() { return *current_children.get().back(); } -const reference_map_t>> &MetaPipeline::GetDependencies() const { +const meta_pipeline_dependency_map_t &MetaPipeline::GetDependencies() const { return pipeline_dependencies; } +bool MetaPipeline::RemoveOptionalDependency(Pipeline &pipeline, Pipeline &dependency) { + auto entry = pipeline_dependencies.find(pipeline); + if (entry == pipeline_dependencies.end()) { + return false; + } + for (auto dependency_entry = entry->second.begin(); dependency_entry != entry->second.end(); dependency_entry++) { + if (dependency_entry->type == MetaPipelineDependencyType::OPTIONAL_DEPENDENCY && + RefersToSameObject(dependency_entry->pipeline.get(), dependency)) { + entry->second.erase(dependency_entry); + return true; + } + } + return false; +} + +void MetaPipeline::AddOptionalDependency(Pipeline &pipeline, Pipeline &dependency) { + pipeline_dependencies[pipeline].emplace_back(dependency, MetaPipelineDependencyType::OPTIONAL_DEPENDENCY); +} + MetaPipelineType MetaPipeline::Type() const { return type; } @@ -119,7 +138,8 @@ Pipeline &MetaPipeline::CreatePipeline() { } vector> MetaPipeline::AddDependenciesFrom(Pipeline &dependant, const Pipeline &start, - const bool including) { + const bool including, + MetaPipelineDependencyType dependency_type) { // find 'start' auto it = pipelines.begin(); for (; !RefersToSameObject(**it, start); it++) { @@ -142,7 +162,7 @@ vector> MetaPipeline::AddDependenciesFrom(Pipeline &dependa // add them to the dependencies auto &explicit_deps = pipeline_dependencies[dependant]; for (auto &created_pipeline : created_pipelines) { - explicit_deps.push_back(*created_pipeline); + explicit_deps.emplace_back(*created_pipeline, dependency_type); } return created_pipelines; @@ -187,10 +207,13 @@ void MetaPipeline::AddRecursiveDependencies(const vector> & !PipelineExceedsThreadCount(*pipeline, thread_count)) { continue; } + auto dependency_type = dependency_mode == RecursiveDependencyMode::FORCE + ? MetaPipelineDependencyType::REQUIRED + : MetaPipelineDependencyType::OPTIONAL_DEPENDENCY; auto &pipeline_deps = pipeline_dependencies[*pipeline]; for (auto &new_dependency : new_dependencies) { if (dataflow_mode == DataflowDependencyMode::SKIP_CONFLICTING) { - bool conflicts_with_dataflow = false; + bool conflicts_with_dataflow = pipeline->HasExternalInputProducer(*new_dependency); for (auto &dataflow_dependency : pipeline->GetDataflowDependencies()) { auto dependency = dataflow_dependency.lock(); D_ASSERT(dependency); @@ -207,7 +230,7 @@ void MetaPipeline::AddRecursiveDependencies(const vector> & !PipelineExceedsThreadCount(*new_dependency, thread_count)) { continue; } - pipeline_deps.push_back(*new_dependency); + pipeline_deps.emplace_back(*new_dependency, dependency_type); } } } @@ -249,15 +272,15 @@ Pipeline &MetaPipeline::CreateUnionPipeline(Pipeline ¤t, bool order_matter // 'union_pipeline' inherits ALL dependencies of 'current' (within this MetaPipeline, and across MetaPipelines) union_pipeline.dependencies = current.dependencies; union_pipeline.dataflow_dependencies = current.dataflow_dependencies; - union_pipeline.external_finish_dependencies = current.external_finish_dependencies; + union_pipeline.input_mode = current.input_mode; + union_pipeline.external_input_producers = current.external_input_producers; auto it = pipeline_dependencies.find(current); if (it != pipeline_dependencies.end()) { pipeline_dependencies[union_pipeline] = it->second; } - if (order_matters) { // if we need to preserve order, or if the sink is not parallel, we set a dependency - pipeline_dependencies[union_pipeline].push_back(current); + pipeline_dependencies[union_pipeline].emplace_back(current, MetaPipelineDependencyType::REQUIRED); } return union_pipeline; @@ -274,7 +297,7 @@ void MetaPipeline::CreateChildPipeline(Pipeline ¤t, PhysicalOperator &op, // child pipeline has a dependency (within this MetaPipeline on all pipelines that were scheduled // between 'current' and now (including 'current') - set them up - pipeline_dependencies[child_pipeline].push_back(current); + pipeline_dependencies[child_pipeline].emplace_back(current, MetaPipelineDependencyType::REQUIRED); AddDependenciesFrom(child_pipeline, last_pipeline, false); D_ASSERT(pipeline_dependencies.find(child_pipeline) != pipeline_dependencies.end()); } diff --git a/src/duckdb/src/parallel/pipeline.cpp b/src/duckdb/src/parallel/pipeline.cpp index be5cd5e7a..921a61d70 100644 --- a/src/duckdb/src/parallel/pipeline.cpp +++ b/src/duckdb/src/parallel/pipeline.cpp @@ -1,4 +1,5 @@ #include "duckdb/parallel/pipeline.hpp" +#include "duckdb/parallel/pipeline_broadcast_exchange.hpp" #include "duckdb/common/algorithm.hpp" #include "duckdb/common/printer.hpp" @@ -7,13 +8,18 @@ #include "duckdb/execution/operator/aggregate/physical_ungrouped_aggregate.hpp" #include "duckdb/execution/operator/helper/physical_result_collector.hpp" #include "duckdb/execution/operator/scan/physical_table_scan.hpp" +#include "duckdb/execution/operator/set/physical_cte.hpp" #include "duckdb/execution/operator/set/physical_recursive_cte.hpp" +#include "duckdb/logging/log_type.hpp" +#include "duckdb/logging/logger.hpp" #include "duckdb/main/client_context.hpp" -#include "duckdb/main/database.hpp" +#include "duckdb/parallel/meta_pipeline.hpp" #include "duckdb/parallel/pipeline_event.hpp" #include "duckdb/parallel/pipeline_executor.hpp" +#include "duckdb/parallel/pipeline_schedule.hpp" #include "duckdb/parallel/task_scheduler.hpp" #include "duckdb/main/settings.hpp" +#include "duckdb/storage/buffer_manager.hpp" namespace duckdb { @@ -215,13 +221,39 @@ bool Pipeline::IsOrderDependent() const { return false; } -void Pipeline::SetExternalInput() { +void Pipeline::SetExternalInput(const vector> &producer_pipelines) { + D_ASSERT(!producer_pipelines.empty()); input_mode = PipelineInputMode::EXTERNAL_INPUT; + external_input_producers.clear(); + external_input_producers.reserve(producer_pipelines.size()); + for (auto &producer : producer_pipelines) { + external_input_producers.emplace_back(producer.get().shared_from_this()); + } annotated_lock_guard guard(external_input_lock); external_input_event.reset(); external_input_event_state = ExternalInputEventState::EXTERNAL_INPUT_UNSET; } +void Pipeline::ClearExternalInput() { + D_ASSERT(IsExternalInput()); + input_mode = PipelineInputMode::SCHEDULED_SOURCE; + external_input_producers.clear(); + annotated_lock_guard guard(external_input_lock); + external_input_event.reset(); + external_input_event_state = ExternalInputEventState::EXTERNAL_INPUT_UNSET; +} + +bool Pipeline::HasExternalInputProducer(const Pipeline &pipeline) const { + for (auto &producer_ref : external_input_producers) { + auto producer = producer_ref.lock(); + D_ASSERT(producer); + if (RefersToSameObject(*producer, pipeline)) { + return true; + } + } + return false; +} + bool Pipeline::IsStreamingResultPipeline() const { if (external_streaming_result_producer) { return true; @@ -263,6 +295,18 @@ bool Pipeline::CanUseExternalInput(const OperatorPartitionInfo &source_partition return true; } +PipelineExternalInputCost Pipeline::GetExternalInputCost() const { + if (sink && sink->GetExternalInputCost() == PipelineExternalInputCost::SERIALIZED_FANOUT) { + return PipelineExternalInputCost::SERIALIZED_FANOUT; + } + for (auto &op_ref : operators) { + if (op_ref.get().GetExternalInputCost() == PipelineExternalInputCost::SERIALIZED_FANOUT) { + return PipelineExternalInputCost::SERIALIZED_FANOUT; + } + } + return PipelineExternalInputCost::PIPELINED; +} + bool Pipeline::CanStopSourceEarly() const { // Used by CTE fanout selection to keep streaming only when the consumer may finish early. if (sink && sink->GetSourceConsumption() == PipelineSourceConsumption::MAY_STOP_EARLY) { @@ -566,15 +610,35 @@ void Pipeline::AddDependency(shared_ptr &pipeline) { pipeline->parents.push_back(weak_ptr(shared_from_this())); } -void Pipeline::AddDataflowDependency(shared_ptr &pipeline) { +void Pipeline::RemoveDependency(const shared_ptr &pipeline) { D_ASSERT(pipeline); - dataflow_dependencies.push_back(weak_ptr(pipeline)); - pipeline->parents.push_back(weak_ptr(shared_from_this())); + bool found_dependency = false; + for (auto entry = dependencies.begin(); entry != dependencies.end(); entry++) { + auto dependency = entry->lock(); + D_ASSERT(dependency); + if (RefersToSameObject(*dependency, *pipeline)) { + dependencies.erase(entry); + found_dependency = true; + break; + } + } + if (!found_dependency) { + throw InternalException("Attempted to remove a missing pipeline dependency"); + } + for (auto entry = pipeline->parents.begin(); entry != pipeline->parents.end(); entry++) { + auto parent = entry->lock(); + D_ASSERT(parent); + if (RefersToSameObject(*parent, *this)) { + pipeline->parents.erase(entry); + return; + } + } + throw InternalException("Pipeline dependency had no matching parent"); } -void Pipeline::AddExternalFinishDependency(shared_ptr &pipeline) { +void Pipeline::AddDataflowDependency(shared_ptr &pipeline) { D_ASSERT(pipeline); - external_finish_dependencies.push_back(weak_ptr(pipeline)); + dataflow_dependencies.push_back(weak_ptr(pipeline)); pipeline->parents.push_back(weak_ptr(shared_from_this())); } @@ -664,6 +728,206 @@ idx_t Pipeline::UpdateBatchIndex(idx_t old_index, idx_t new_index) { //===--------------------------------------------------------------------===// // Pipeline Build State //===--------------------------------------------------------------------===// +struct PipelineExternalInputCandidate { + enum class InputMode : uint8_t { MATERIALIZED_COST, EXTERNAL_INPUT, MATERIALIZED_CYCLE }; + + PipelineExternalInputCandidate(Pipeline &pipeline_p, PhysicalOperator &materialized_source_p, + PhysicalOperator &external_source_p, shared_ptr cte_dependency_p, + PhysicalCTE &cte_p, idx_t consumer_idx_p) + : pipeline(pipeline_p), materialized_source(materialized_source_p), external_source(external_source_p), + cte_dependency(std::move(cte_dependency_p)), cte(cte_p), consumer_idx(consumer_idx_p), + has_blocking_dependencies(pipeline_p.GetDependencies().size() > 1), + input_cost(pipeline_p.GetExternalInputCost()) { + } + + reference pipeline; + reference materialized_source; + reference external_source; + shared_ptr cte_dependency; + reference cte; + idx_t consumer_idx; + idx_t selection_idx = DConstants::INVALID_INDEX; + bool has_blocking_dependencies; + PipelineExternalInputCost input_cost; + InputMode input_mode = InputMode::MATERIALIZED_COST; + + bool UsesExternalInput() const { + return input_mode == InputMode::EXTERNAL_INPUT; + } +}; + +struct CTEPipelineSelection { + CTEPipelineSelection(PhysicalCTE &cte_p, Pipeline &pipeline_p, shared_ptr cte_dependency_p, + bool dependency_added_p) + : cte(cte_p), pipeline(pipeline_p), cte_dependency(std::move(cte_dependency_p)), + fallback_dependency_added(dependency_added_p), fallback_dependency_active(dependency_added_p) { + } + + reference cte; + reference pipeline; + shared_ptr cte_dependency; + idx_t candidate_count = 0; + idx_t serialized_fanout_candidate_count = 0; + idx_t direct_candidate_count = 0; + idx_t estimated_materialization_size = 0; + bool fallback_dependency_added; + bool fallback_dependency_active; + bool stream_serialized_candidates = false; +}; + +class PipelineBuildStateData { +public: + vector external_input_candidates; + vector cte_selections; + reference_map_t cte_selection_map; +}; + +struct RemovedOptionalPipelineDependency { + RemovedOptionalPipelineDependency(MetaPipeline &meta_pipeline_p, Pipeline &pipeline_p, Pipeline &dependency_p) + : meta_pipeline(meta_pipeline_p), pipeline(pipeline_p), dependency(dependency_p) { + } + + reference meta_pipeline; + reference pipeline; + reference dependency; +}; + +static bool RemoveOptionalDependencyInCycle(const PipelineSchedule &schedule, const vector &cycle, + const vector> &meta_pipelines, + vector &removed_dependencies) { + for (auto &edge : cycle) { + auto &pipeline_stage = schedule.stages[edge.dependent]; + auto &dependency_stage = schedule.stages[edge.dependency]; + if (pipeline_stage.type != PipelineScheduleStageType::EXECUTE || + dependency_stage.type != PipelineScheduleStageType::EXECUTE) { + continue; + } + for (auto &meta_pipeline : meta_pipelines) { + if (meta_pipeline->RemoveOptionalDependency(*pipeline_stage.pipeline, *dependency_stage.pipeline)) { + removed_dependencies.emplace_back(*meta_pipeline, *pipeline_stage.pipeline, *dependency_stage.pipeline); + return true; + } + } + } + return false; +} + +static void RestoreOptionalDependencies(vector &dependencies) { + for (auto &dependency : dependencies) { + dependency.meta_pipeline.get().AddOptionalDependency(dependency.pipeline, dependency.dependency); + } + dependencies.clear(); +} + +static idx_t SaturatingAdd(idx_t left, idx_t right) { + auto maximum = NumericLimits::Maximum(); + return right > maximum - left ? maximum : left + right; +} + +static idx_t EstimateMaterializationSize(const PhysicalCTE &cte) { + D_ASSERT(!cte.children.empty()); + auto &producer = cte.children[0].get(); + idx_t row_width = 0; + for (auto &type : producer.GetTypes()) { + row_width = SaturatingAdd(row_width, MaxValue(GetTypeIdSize(type.InternalType()), 1)); + } + if (row_width == 0 || producer.estimated_cardinality == 0) { + return 0; + } + auto maximum = NumericLimits::Maximum(); + return producer.estimated_cardinality > maximum / row_width ? maximum : producer.estimated_cardinality * row_width; +} + +static void SelectExternalInputCandidates(PipelineBuildStateData &data) { + for (auto &candidate : data.external_input_candidates) { + auto selection_entry = data.cte_selection_map.find(candidate.cte.get()); + D_ASSERT(selection_entry != data.cte_selection_map.end()); + candidate.selection_idx = selection_entry->second; + auto &selection = data.cte_selections[candidate.selection_idx]; + selection.candidate_count++; + if (candidate.input_cost == PipelineExternalInputCost::SERIALIZED_FANOUT) { + selection.serialized_fanout_candidate_count++; + } + } + + vector fanout_selections; + for (idx_t selection_idx = 0; selection_idx < data.cte_selections.size(); selection_idx++) { + auto &selection = data.cte_selections[selection_idx]; + auto &cte = selection.cte.get(); + D_ASSERT(cte.exchange); + auto consumer_summary = cte.exchange->GetConsumerSummary(); + D_ASSERT(consumer_summary.unresolved == selection.candidate_count); + if (selection.serialized_fanout_candidate_count == 0) { + continue; + } + // An existing materialized consumer already makes the materialization cost unavoidable. + if (consumer_summary.materialized > 0) { + continue; + } + // A single consumer cannot introduce serialized fanout. + if (selection.candidate_count == 1 && consumer_summary.ExchangeConsumerCount() == 0) { + selection.stream_serialized_candidates = true; + continue; + } + selection.estimated_materialization_size = EstimateMaterializationSize(cte); + fanout_selections.push_back(selection_idx); + } + + // Direct fanout trades materialization memory for serialized pushes into every consumer. Keep small fanouts + // materialized, and stream the largest CTEs when materializing all eligible inputs would consume too much memory. + auto &context = data.external_input_candidates[0].pipeline.get().GetClientContext(); + auto materialization_budget = BufferManager::GetBufferManager(context).GetOperatorMemoryLimit() / 2; + std::sort(fanout_selections.begin(), fanout_selections.end(), [&](idx_t left, idx_t right) { + return data.cte_selections[left].estimated_materialization_size < + data.cte_selections[right].estimated_materialization_size; + }); + idx_t materialization_size = 0; + for (auto selection_idx : fanout_selections) { + auto &selection = data.cte_selections[selection_idx]; + if (selection.estimated_materialization_size <= materialization_budget - materialization_size) { + materialization_size += selection.estimated_materialization_size; + } else { + selection.stream_serialized_candidates = true; + } + } + + for (auto &candidate : data.external_input_candidates) { + auto &selection = data.cte_selections[candidate.selection_idx]; + if (candidate.input_cost == PipelineExternalInputCost::PIPELINED || selection.stream_serialized_candidates) { + candidate.input_mode = PipelineExternalInputCandidate::InputMode::EXTERNAL_INPUT; + } + } +} + +static optional_ptr +FindExternalInputCandidateInCycle(PipelineBuildStateData &data, const vector &cycle) { + reference_set_t cycle_consumers; + for (auto &edge : cycle) { + if (edge.external_input_consumer) { + cycle_consumers.insert(*edge.external_input_consumer); + } + } + optional_ptr result; + for (auto &candidate : data.external_input_candidates) { + if (!candidate.UsesExternalInput()) { + continue; + } + if (cycle_consumers.find(candidate.pipeline.get()) == cycle_consumers.end()) { + continue; + } + if (candidate.has_blocking_dependencies) { + return candidate; + } + result = candidate; + } + return result; +} + +PipelineBuildState::PipelineBuildState() : data(make_uniq()) { +} + +PipelineBuildState::~PipelineBuildState() = default; + void PipelineBuildState::SetPipelineSource(Pipeline &pipeline, PhysicalOperator &op) { pipeline.source = &op; } @@ -700,4 +964,104 @@ vector> PipelineBuildState::GetPipelineOperators(Pip return pipeline.operators; } +void PipelineBuildState::AddExternalInputCandidate(Pipeline &pipeline, PhysicalOperator &materialized_source, + PhysicalOperator &external_source, + shared_ptr cte_dependency, PhysicalCTE &cte, + idx_t consumer_idx) { + data->external_input_candidates.emplace_back(pipeline, materialized_source, external_source, + std::move(cte_dependency), cte, consumer_idx); +} + +void PipelineBuildState::AddCTEPipelineSelection(PhysicalCTE &cte, Pipeline &pipeline, + shared_ptr cte_dependency, bool dependency_added) { + D_ASSERT(data->cte_selection_map.find(cte) == data->cte_selection_map.end()); + data->cte_selection_map.emplace(cte, data->cte_selections.size()); + data->cte_selections.emplace_back(cte, pipeline, std::move(cte_dependency), dependency_added); +} + +void PipelineBuildState::ResolveExternalInputs(const vector> &meta_pipelines) { + if (data->external_input_candidates.empty()) { + return; + } + SelectExternalInputCandidates(*data); + auto activate_candidate = [&](PipelineExternalInputCandidate &candidate) { + D_ASSERT(candidate.UsesExternalInput()); + auto &pipeline = candidate.pipeline.get(); + pipeline.RemoveDependency(candidate.cte_dependency); + SetPipelineSource(pipeline, candidate.external_source.get()); + pipeline.SetExternalInput(candidate.cte.get().GetProducerPipelines()); + data->cte_selections[candidate.selection_idx].direct_candidate_count++; + }; + auto restore_materialized_candidate = [&](PipelineExternalInputCandidate &candidate) { + D_ASSERT(candidate.UsesExternalInput()); + auto &pipeline = candidate.pipeline.get(); + pipeline.ClearExternalInput(); + SetPipelineSource(pipeline, candidate.materialized_source.get()); + pipeline.AddDependency(candidate.cte_dependency); + candidate.input_mode = PipelineExternalInputCandidate::InputMode::MATERIALIZED_CYCLE; + + auto &selection = data->cte_selections[candidate.selection_idx]; + D_ASSERT(selection.direct_candidate_count > 0); + selection.direct_candidate_count--; + if (selection.direct_candidate_count == 0 && selection.fallback_dependency_added && + !selection.fallback_dependency_active) { + selection.pipeline.get().AddDependency(selection.cte_dependency); + selection.fallback_dependency_active = true; + } + }; + for (auto &candidate : data->external_input_candidates) { + if (candidate.UsesExternalInput()) { + activate_candidate(candidate); + } + } + for (auto &selection : data->cte_selections) { + if (selection.direct_candidate_count == 0 || !selection.fallback_dependency_active) { + continue; + } + selection.pipeline.get().RemoveDependency(selection.cte_dependency); + selection.fallback_dependency_active = false; + } + + vector removed_dependencies; + while (true) { + auto schedule = BuildPipelineSchedule(meta_pipelines); + auto cycle = schedule->GetCycle(); + if (cycle.empty()) { + break; + } + if (RemoveOptionalDependencyInCycle(*schedule, cycle, meta_pipelines, removed_dependencies)) { + continue; + } + auto candidate = FindExternalInputCandidateInCycle(*data, cycle); + if (!candidate) { + throw InternalException("Cyclic dependency in pipeline schedule without an external input candidate"); + } + restore_materialized_candidate(*candidate); + RestoreOptionalDependencies(removed_dependencies); + } + + for (auto &candidate : data->external_input_candidates) { + auto &pipeline = candidate.pipeline.get(); + auto &cte = candidate.cte.get(); + if (candidate.UsesExternalInput()) { + cte.RegisterDirectConsumer(pipeline, candidate.consumer_idx); + DUCKDB_LOG(pipeline.GetClientContext(), PhysicalOperatorLogType, cte, "PhysicalCTE", "SelectConsumer", + {{"consumer", to_string(candidate.consumer_idx)}, {"mode", "DIRECT"}}); + } else { + cte.RegisterMaterializedConsumer(candidate.consumer_idx); + DUCKDB_LOG(pipeline.GetClientContext(), PhysicalOperatorLogType, cte, "PhysicalCTE", "SelectConsumer", + {{"consumer", to_string(candidate.consumer_idx)}, + {"mode", "MATERIALIZED"}, + {"reason", candidate.input_mode == PipelineExternalInputCandidate::InputMode::MATERIALIZED_CYCLE + ? "CYCLE" + : "COST"}}); + } + } + + for (auto &selection : data->cte_selections) { + selection.cte.get().SetPipelineSelectionResolved(); + } + D_ASSERT(!BuildPipelineSchedule(meta_pipelines)->HasCycle()); +} + } // namespace duckdb diff --git a/src/duckdb/src/parallel/pipeline_broadcast_exchange.cpp b/src/duckdb/src/parallel/pipeline_broadcast_exchange.cpp index cdb0d817d..9465fe588 100644 --- a/src/duckdb/src/parallel/pipeline_broadcast_exchange.cpp +++ b/src/duckdb/src/parallel/pipeline_broadcast_exchange.cpp @@ -601,6 +601,9 @@ PipelineBroadcastExchangeLocalState::PipelineBroadcastExchangeLocalState(ClientC auto &pipeline = pipeline_ref.get(); pipeline.PrepareExternalInput(); direct_executors.push_back(make_uniq(context, pipeline)); + auto input_chunk = make_uniq(); + input_chunk->InitializeEmpty(exchange.Types()); + direct_input_chunks.push_back(std::move(input_chunk)); } } @@ -661,7 +664,7 @@ void PipelineBroadcastExchange::SelectMaterializedConsumer(idx_t consumer_idx) { DeactivateConsumerLocked(consumer, buffer->NextPosition()); } -bool PipelineBroadcastExchange::TryRegisterDirectConsumer(Pipeline &pipeline, idx_t consumer_idx) { +bool PipelineBroadcastExchange::CanRegisterDirectConsumer(Pipeline &pipeline) const { auto source_partition_info = SupportsBatchIndex() ? OperatorPartitionInfo::BatchIndex() : OperatorPartitionInfo::NoPartitionInfo(); if (!pipeline.CanUseExternalInput(source_partition_info)) { @@ -672,10 +675,15 @@ bool PipelineBroadcastExchange::TryRegisterDirectConsumer(Pipeline &pipeline, id if (required_partition_info.RequiresBatchIndex() && producer_pipelines.size() != 1) { return false; } + return true; +} + +void PipelineBroadcastExchange::SelectDirectConsumer(Pipeline &pipeline, idx_t consumer_idx) { + annotated_lock_guard guard(lock); D_ASSERT(consumer_idx < consumers.size()); auto &consumer = consumers[consumer_idx]; if (consumer.mode == PipelineBroadcastExchangeConsumerMode::DIRECT) { - return true; + return; } D_ASSERT(consumer.mode == PipelineBroadcastExchangeConsumerMode::UNRESOLVED); consumer.mode = PipelineBroadcastExchangeConsumerMode::DIRECT; @@ -686,9 +694,22 @@ bool PipelineBroadcastExchange::TryRegisterDirectConsumer(Pipeline &pipeline, id producer_pipeline.get().SetExternalStreamingResultProducer(); } } +} + +bool PipelineBroadcastExchange::TryRegisterDirectConsumer(Pipeline &pipeline, idx_t consumer_idx) { + if (!CanRegisterDirectConsumer(pipeline)) { + return false; + } + pipeline.SetExternalInput(GetProducerPipelines()); + SelectDirectConsumer(pipeline, consumer_idx); return true; } +vector> PipelineBroadcastExchange::GetProducerPipelines() const { + annotated_lock_guard guard(lock); + return producer_pipelines; +} + void PipelineBroadcastExchange::SelectBufferedConsumer(idx_t consumer_idx, PipelineBroadcastExchangeScanMode scan_mode) { annotated_lock_guard guard(lock); @@ -877,7 +898,8 @@ SinkCombineResultType PipelineBroadcastExchange::FinishLocal(PipelineBroadcastEx SinkResultType PipelineBroadcastExchangeLocalState::Push(DataChunk &chunk, const SourcePartitionInfo &partition_info, const InterruptState &interrupt_state) { - if (direct_push_state != PipelineBroadcastExchangeDirectPushState::RESUMING) { + auto resuming = direct_push_state == PipelineBroadcastExchangeDirectPushState::RESUMING; + if (!resuming) { direct_idx = 0; } auto source_partition_data = GetSourcePartitionData(partition_info); @@ -886,13 +908,20 @@ SinkResultType PipelineBroadcastExchangeLocalState::Push(DataChunk &chunk, const auto &executor = *direct_executors[direct_idx]; executor.SetInterruptState(interrupt_state); if (executor.IsFinishedProcessing()) { + resuming = false; continue; } - auto result = executor.PushExternal(chunk, source_partition_data, source_min_batch_index); + auto &input_chunk = *direct_input_chunks[direct_idx]; + if (!resuming) { + // Operators may slice their input chunk. Keep each direct consumer's wrapper independent. + input_chunk.Reference(chunk); + } + auto result = executor.PushExternal(input_chunk, source_partition_data, source_min_batch_index); if (result == PipelineExecuteResult::INTERRUPTED) { direct_push_state = PipelineBroadcastExchangeDirectPushState::RESUMING; return SinkResultType::BLOCKED; } + resuming = false; } direct_push_state = PipelineBroadcastExchangeDirectPushState::FINISHED; diff --git a/src/duckdb/src/parallel/pipeline_schedule.cpp b/src/duckdb/src/parallel/pipeline_schedule.cpp index 5d1741cae..a8466482a 100644 --- a/src/duckdb/src/parallel/pipeline_schedule.cpp +++ b/src/duckdb/src/parallel/pipeline_schedule.cpp @@ -34,6 +34,77 @@ static void AddDependency(PipelineSchedule &result, idx_t dependent, idx_t depen result.stages[dependent].dependencies.push_back(dependency); } +static void AddExternalInputDependency(PipelineSchedule &result, idx_t dependent, idx_t dependency, + Pipeline &consumer) { + auto dependency_index = result.stages[dependent].dependencies.size(); + result.stages[dependent].dependencies.push_back(dependency); + result.external_input_dependencies.emplace_back(dependent, dependency_index, consumer); +} + +static optional_ptr GetExternalInputConsumer(const PipelineSchedule &schedule, idx_t dependent, + idx_t dependency_index) { + for (auto &dependency : schedule.external_input_dependencies) { + if (dependency.dependent == dependent && dependency.dependency_index == dependency_index) { + return dependency.consumer.get(); + } + } + return nullptr; +} + +enum class CycleVisitState : uint8_t { UNVISITED, VISITING, VISITED }; + +vector PipelineSchedule::GetCycle() const { + vector states(stages.size(), CycleVisitState::UNVISITED); + for (idx_t stage_idx = 0; stage_idx < stages.size(); stage_idx++) { + if (states[stage_idx] != CycleVisitState::UNVISITED) { + continue; + } + vector> stack; + stack.emplace_back(stage_idx, 0); + states[stage_idx] = CycleVisitState::VISITING; + while (!stack.empty()) { + auto &entry = stack.back(); + auto &dependencies = stages[entry.first].dependencies; + if (entry.second == dependencies.size()) { + states[entry.first] = CycleVisitState::VISITED; + stack.pop_back(); + continue; + } + auto dependency = dependencies[entry.second++]; + D_ASSERT(dependency < stages.size()); + if (states[dependency] == CycleVisitState::UNVISITED) { + states[dependency] = CycleVisitState::VISITING; + stack.emplace_back(dependency, 0); + continue; + } + if (states[dependency] != CycleVisitState::VISITING) { + continue; + } + vector result; + idx_t cycle_start = 0; + while (stack[cycle_start].first != dependency) { + cycle_start++; + D_ASSERT(cycle_start < stack.size()); + } + for (idx_t path_idx = cycle_start; path_idx + 1 < stack.size(); path_idx++) { + auto dependent = stack[path_idx].first; + auto dependency_idx = stack[path_idx].second - 1; + result.emplace_back(dependent, stages[dependent].dependencies[dependency_idx], + GetExternalInputConsumer(*this, dependent, dependency_idx)); + } + auto dependent = stack.back().first; + auto dependency_idx = stack.back().second - 1; + result.emplace_back(dependent, dependency, GetExternalInputConsumer(*this, dependent, dependency_idx)); + return result; + } + } + return {}; +} + +bool PipelineSchedule::HasCycle() const { + return !GetCycle().empty(); +} + static PipelineScheduleStageStack AddBasePipeline(PipelineSchedule &result, const shared_ptr &pipeline) { auto initialize = AddStage(result, PipelineScheduleStageType::INITIALIZE, pipeline); auto execute = AddStage(result, PipelineScheduleStageType::EXECUTE, pipeline); @@ -142,16 +213,6 @@ unique_ptr BuildPipelineSchedule(const vectorsecond.initialize); } } - // External finish dependencies block both execution and PrepareFinalize on the producer's execution. - for (auto &dependency : pipeline.GetExternalFinishDependencies()) { - auto dep = dependency.lock(); - D_ASSERT(dep); - auto dep_entry = stage_map.find(*dep); - if (dep_entry != stage_map.end()) { - AddDependency(*result, entry.second.execute, dep_entry->second.execute); - AddDependency(*result, entry.second.prepare_finish, dep_entry->second.execute); - } - } } // Meta-pipeline dependencies order their execute stages directly. @@ -165,7 +226,7 @@ unique_ptr BuildPipelineSchedule(const vector BuildPipelineSchedule(const vectorsecond.execute, entry.second.initialize, consumer); + for (auto &dependency : consumer.GetDependencies()) { + auto dep = dependency.lock(); + D_ASSERT(dep); + auto dep_entry = stage_map.find(*dep); + if (dep_entry != stage_map.end()) { + AddExternalInputDependency(*result, producer_entry->second.execute, dep_entry->second.complete, + consumer); + } + } + for (auto &dependency : consumer.GetDataflowDependencies()) { + auto dep = dependency.lock(); + D_ASSERT(dep); + auto dep_entry = stage_map.find(*dep); + if (dep_entry != stage_map.end()) { + AddExternalInputDependency(*result, producer_entry->second.execute, dep_entry->second.initialize, + consumer); + } + } + AddExternalInputDependency(*result, entry.second.execute, producer_entry->second.execute, consumer); + AddExternalInputDependency(*result, entry.second.prepare_finish, producer_entry->second.execute, consumer); + } + } + return result; } diff --git a/src/duckdb/src/storage/table/column_data_checkpointer.cpp b/src/duckdb/src/storage/table/column_data_checkpointer.cpp index 67a7b8323..ebd2d07c8 100644 --- a/src/duckdb/src/storage/table/column_data_checkpointer.cpp +++ b/src/duckdb/src/storage/table/column_data_checkpointer.cpp @@ -190,32 +190,25 @@ vector ColumnDataCheckpointer::DetectBestCompressionMet } InitAnalyze(); - - // If the compression type was explicitly specified at column definition time, - // the decision is already made — skip the entire analyze scan. - const bool skip_scan = (compression_type != CompressionType::COMPRESSION_AUTO); - - if (!skip_scan) { - // scan over all the segments and run the analyze step - ScanSegments([&](Vector &scan_vector) { - for (idx_t i = 0; i < checkpoint_states.size(); i++) { - auto &functions = compression_functions[i]; - auto &states = analyze_states[i]; - for (idx_t j = 0; j < functions.size(); j++) { - auto &state = states[j]; - auto &func = functions[j]; - - if (!state) { - continue; - } - if (!func->analyze(*state, scan_vector)) { - state = nullptr; - func = nullptr; - } + // scan over all the segments and run the analyze step + ScanSegments([&](Vector &scan_vector) { + for (idx_t i = 0; i < checkpoint_states.size(); i++) { + auto &functions = compression_functions[i]; + auto &states = analyze_states[i]; + for (idx_t j = 0; j < functions.size(); j++) { + auto &state = states[j]; + auto &func = functions[j]; + + if (!state) { + continue; + } + if (!func->analyze(*state, scan_vector)) { + state = nullptr; + func = nullptr; } } - }); - } + } + }); vector result; result.resize(checkpoint_states.size());