| /* |
| * Copyright (C) 2018 The Android Open Source Project |
| * |
| * Licensed under the Apache License, Version 2.0 (the "License"); |
| * you may not use this file except in compliance with the License. |
| * You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| #include "src/trace_processor/span_join_operator_table.h" |
| |
| #include <sqlite3.h> |
| #include <string.h> |
| #include <algorithm> |
| #include <set> |
| #include <utility> |
| |
| #include "perfetto/base/logging.h" |
| #include "perfetto/base/string_splitter.h" |
| #include "perfetto/base/string_utils.h" |
| #include "perfetto/base/string_view.h" |
| #include "src/trace_processor/sqlite_utils.h" |
| |
| namespace perfetto { |
| namespace trace_processor { |
| |
| namespace { |
| |
| constexpr char kTsColumnName[] = "ts"; |
| constexpr char kDurColumnName[] = "dur"; |
| |
| bool IsRequiredColumn(const std::string& name) { |
| return name == kTsColumnName || name == kDurColumnName; |
| } |
| |
| } // namespace |
| |
| SpanJoinOperatorTable::SpanJoinOperatorTable(sqlite3* db, const TraceStorage*) |
| : db_(db) {} |
| |
| void SpanJoinOperatorTable::RegisterTable(sqlite3* db, |
| const TraceStorage* storage) { |
| Table::Register<SpanJoinOperatorTable>(db, storage, "span_join", |
| /* read_write */ false, |
| /* requires_args */ true); |
| |
| Table::Register<SpanJoinOperatorTable>(db, storage, "span_left_join", |
| /* read_write */ false, |
| /* requires_args */ true); |
| |
| Table::Register<SpanJoinOperatorTable>(db, storage, "span_outer_join", |
| /* read_write */ false, |
| /* requires_args */ true); |
| } |
| |
| base::Optional<Table::Schema> SpanJoinOperatorTable::Init( |
| int argc, |
| const char* const* argv) { |
| // argv[0] - argv[2] are SQLite populated fields which are always present. |
| if (argc < 5) { |
| PERFETTO_ELOG("SPAN JOIN expected at least 2 args, received %d", argc - 3); |
| return base::nullopt; |
| } |
| |
| auto maybe_t1_desc = TableDescriptor::Parse( |
| std::string(reinterpret_cast<const char*>(argv[3]))); |
| if (!maybe_t1_desc.has_value()) |
| return base::nullopt; |
| auto t1_desc = *maybe_t1_desc; |
| |
| auto maybe_t2_desc = TableDescriptor::Parse( |
| std::string(reinterpret_cast<const char*>(argv[4]))); |
| if (!maybe_t2_desc.has_value()) |
| return base::nullopt; |
| auto t2_desc = *maybe_t2_desc; |
| |
| if (t1_desc.partition_col == t2_desc.partition_col) { |
| partitioning_ = t1_desc.IsPartitioned() |
| ? PartitioningType::kSamePartitioning |
| : PartitioningType::kNoPartitioning; |
| if (partitioning_ == PartitioningType::kNoPartitioning && IsOuterJoin()) { |
| PERFETTO_ELOG("Outer join not supported for no partition tables"); |
| return base::nullopt; |
| } |
| } else if (t1_desc.IsPartitioned() && t2_desc.IsPartitioned()) { |
| PERFETTO_ELOG("Mismatching partitions (%s, %s)", |
| t1_desc.partition_col.c_str(), t2_desc.partition_col.c_str()); |
| return base::nullopt; |
| } else { |
| if (IsOuterJoin()) { |
| PERFETTO_ELOG("Outer join not supported for mixed partitioned tables"); |
| return base::nullopt; |
| } |
| partitioning_ = PartitioningType::kMixedPartitioning; |
| } |
| |
| auto maybe_t1_defn = CreateTableDefinition(t1_desc, IsOuterJoin()); |
| if (!maybe_t1_defn.has_value()) |
| return base::nullopt; |
| t1_defn_ = maybe_t1_defn.value(); |
| |
| auto maybe_t2_defn = |
| CreateTableDefinition(t2_desc, IsOuterJoin() || IsLeftJoin()); |
| if (!maybe_t2_defn.has_value()) |
| return base::nullopt; |
| t2_defn_ = maybe_t2_defn.value(); |
| |
| std::vector<Table::Column> cols; |
| // Ensure the shared columns are consistently ordered and are not |
| // present twice in the final schema |
| cols.emplace_back(Column::kTimestamp, kTsColumnName, ColumnType::kLong); |
| cols.emplace_back(Column::kDuration, kDurColumnName, ColumnType::kLong); |
| if (partitioning_ != PartitioningType::kNoPartitioning) |
| cols.emplace_back(Column::kPartition, partition_col(), ColumnType::kLong); |
| |
| CreateSchemaColsForDefn(t1_defn_, &cols); |
| CreateSchemaColsForDefn(t2_defn_, &cols); |
| |
| std::vector<size_t> primary_keys = {Column::kTimestamp}; |
| if (partitioning_ != PartitioningType::kNoPartitioning) |
| primary_keys.push_back(Column::kPartition); |
| return Schema(cols, primary_keys); |
| } |
| |
| void SpanJoinOperatorTable::CreateSchemaColsForDefn( |
| const TableDefinition& defn, |
| std::vector<Table::Column>* cols) { |
| for (size_t i = 0; i < defn.columns().size(); i++) { |
| const auto& n = defn.columns()[i].name(); |
| if (IsRequiredColumn(n) || n == defn.partition_col()) |
| continue; |
| |
| ColumnLocator* locator = &global_index_to_column_locator_[cols->size()]; |
| locator->defn = &defn; |
| locator->col_index = i; |
| |
| cols->emplace_back(cols->size(), n, defn.columns()[i].type()); |
| } |
| } |
| |
| std::unique_ptr<Table::Cursor> SpanJoinOperatorTable::CreateCursor( |
| const QueryConstraints& qc, |
| sqlite3_value** argv) { |
| auto cursor = |
| std::unique_ptr<SpanJoinOperatorTable::Cursor>(new Cursor(this, db_)); |
| int value = cursor->Initialize(qc, argv); |
| return value != SQLITE_OK ? nullptr : std::move(cursor); |
| } |
| |
| int SpanJoinOperatorTable::BestIndex(const QueryConstraints&, BestIndexInfo*) { |
| // TODO(lalitm): figure out cost estimation. |
| return SQLITE_OK; |
| } |
| |
| std::vector<std::string> |
| SpanJoinOperatorTable::ComputeSqlConstraintsForDefinition( |
| const TableDefinition& defn, |
| const QueryConstraints& qc, |
| sqlite3_value** argv) { |
| std::vector<std::string> constraints; |
| for (size_t i = 0; i < qc.constraints().size(); i++) { |
| const auto& cs = qc.constraints()[i]; |
| auto col_name = GetNameForGlobalColumnIndex(defn, cs.iColumn); |
| if (col_name == "") |
| continue; |
| |
| if (col_name == kTsColumnName || col_name == kDurColumnName) { |
| // We don't support constraints on ts or duration in the child tables. |
| PERFETTO_DFATAL("ts or duration constraints on child tables"); |
| continue; |
| } |
| auto op = sqlite_utils::OpToString(cs.op); |
| auto value = sqlite_utils::SqliteValueAsString(argv[i]); |
| |
| constraints.emplace_back("`" + col_name + "`" + op + value); |
| } |
| return constraints; |
| } |
| |
| base::Optional<SpanJoinOperatorTable::TableDefinition> |
| SpanJoinOperatorTable::CreateTableDefinition(const TableDescriptor& desc, |
| bool emit_shadow_slices) { |
| auto cols = sqlite_utils::GetColumnsForTable(db_, desc.name); |
| |
| uint32_t required_columns_found = 0; |
| uint32_t ts_idx = std::numeric_limits<uint32_t>::max(); |
| uint32_t dur_idx = std::numeric_limits<uint32_t>::max(); |
| uint32_t partition_idx = std::numeric_limits<uint32_t>::max(); |
| for (uint32_t i = 0; i < cols.size(); i++) { |
| auto col = cols[i]; |
| if (IsRequiredColumn(col.name())) { |
| ++required_columns_found; |
| if (col.type() != Table::ColumnType::kLong && |
| col.type() != Table::ColumnType::kUnknown) { |
| PERFETTO_ELOG("Invalid column type for %s", col.name().c_str()); |
| return base::nullopt; |
| } |
| } |
| |
| if (col.name() == kTsColumnName) { |
| ts_idx = i; |
| } else if (col.name() == kDurColumnName) { |
| dur_idx = i; |
| } else if (col.name() == desc.partition_col) { |
| partition_idx = i; |
| } |
| } |
| if (required_columns_found != 2) { |
| PERFETTO_ELOG("Required columns not found (found %d)", |
| required_columns_found); |
| return base::nullopt; |
| } |
| |
| PERFETTO_DCHECK(ts_idx < cols.size()); |
| PERFETTO_DCHECK(dur_idx < cols.size()); |
| PERFETTO_DCHECK(desc.partition_col.empty() || partition_idx < cols.size()); |
| |
| return TableDefinition(desc.name, desc.partition_col, std::move(cols), |
| emit_shadow_slices, ts_idx, dur_idx, partition_idx); |
| } |
| |
| std::string SpanJoinOperatorTable::GetNameForGlobalColumnIndex( |
| const TableDefinition& defn, |
| int global_column) { |
| size_t col_idx = static_cast<size_t>(global_column); |
| if (col_idx == Column::kTimestamp) |
| return kTsColumnName; |
| else if (col_idx == Column::kDuration) |
| return kDurColumnName; |
| else if (col_idx == Column::kPartition && |
| partitioning_ != PartitioningType::kNoPartitioning) |
| return defn.partition_col().c_str(); |
| |
| const auto& locator = global_index_to_column_locator_[col_idx]; |
| if (locator.defn != &defn) |
| return ""; |
| return defn.columns()[locator.col_index].name().c_str(); |
| } |
| |
| SpanJoinOperatorTable::Cursor::Cursor(SpanJoinOperatorTable* table, sqlite3* db) |
| : t1_(table, &table->t1_defn_, db), |
| t2_(table, &table->t2_defn_, db), |
| table_(table) {} |
| |
| int SpanJoinOperatorTable::Cursor::Initialize(const QueryConstraints& qc, |
| sqlite3_value** argv) { |
| int err = t1_.Initialize(qc, argv); |
| if (err != SQLITE_OK) |
| return err; |
| |
| err = t2_.Initialize(qc, argv); |
| if (err != SQLITE_OK) |
| return err; |
| |
| // Step the partitioned table to allow for us to look into it below. |
| Query* step_now = t1_.IsPartitioned() ? &t1_ : &t2_; |
| next_stepped_ = step_now == &t1_ ? &t2_ : &t1_; |
| |
| auto res = step_now->Step(); |
| if (PERFETTO_UNLIKELY(res.is_err())) |
| return res.err_code; |
| |
| // Forward the unpartitioned table to reflect the partition of the partitoined |
| // table. |
| if (table_->partitioning_ == PartitioningType::kMixedPartitioning) { |
| PERFETTO_DCHECK(step_now->IsPartitioned()); |
| |
| // If we emit shadow slices, we need to step because the first slice will |
| // be a full partition shadow slice that we need to skip. |
| if (step_now->definition()->emit_shadow_slices()) { |
| PERFETTO_DCHECK(step_now->IsFullPartitionShadowSlice()); |
| res = step_now->StepToNextPartition(); |
| if (PERFETTO_UNLIKELY(res.is_err())) |
| return res.err_code; |
| } |
| |
| res = next_stepped_->StepToPartition(step_now->partition()); |
| if (PERFETTO_UNLIKELY(res.is_err())) |
| return res.err_code; |
| } |
| |
| // Otherwise, find an overlapping span. |
| return Next(); |
| } |
| |
| bool SpanJoinOperatorTable::Cursor::IsOverlappingSpan() { |
| if (!t1_.IsRealSlice() && !t2_.IsRealSlice()) { |
| return false; |
| } else if (t1_.partition() != t2_.partition()) { |
| return false; |
| } else if (t1_.ts_end() <= t2_.ts_start() || t2_.ts_end() <= t1_.ts_start()) { |
| return false; |
| } |
| return true; |
| } |
| |
| int SpanJoinOperatorTable::Cursor::Next() { |
| // TODO: Propagate error msg to the table. |
| auto res = next_stepped_->Step(); |
| if (res.is_err()) |
| return res.err_code; |
| |
| while (true) { |
| if (t1_.Eof() || t2_.Eof()) { |
| if (table_->partitioning_ != PartitioningType::kMixedPartitioning) |
| return SQLITE_OK; |
| |
| auto* partitioned = t1_.IsPartitioned() ? &t1_ : &t2_; |
| auto* unpartitioned = t1_.IsPartitioned() ? &t2_ : &t1_; |
| if (partitioned->Eof()) |
| return SQLITE_OK; |
| |
| res = partitioned->StepToNextPartition(); |
| if (PERFETTO_UNLIKELY(res.is_err())) |
| return res.err_code; |
| else if (PERFETTO_UNLIKELY(res.is_eof())) |
| continue; |
| |
| res = unpartitioned->StepToPartition(partitioned->partition()); |
| if (PERFETTO_UNLIKELY(res.is_err())) |
| return res.err_code; |
| else if (PERFETTO_UNLIKELY(res.is_eof())) |
| continue; |
| } |
| |
| int64_t partition = std::max(t1_.partition(), t2_.partition()); |
| res = t1_.StepToPartition(partition); |
| if (PERFETTO_UNLIKELY(res.is_err())) |
| return res.err_code; |
| else if (PERFETTO_UNLIKELY(res.is_eof())) |
| continue; |
| |
| res = t2_.StepToPartition(t1_.partition()); |
| if (PERFETTO_UNLIKELY(res.is_err())) |
| return res.err_code; |
| else if (PERFETTO_UNLIKELY(res.is_eof())) |
| continue; |
| |
| if (t1_.partition() != t2_.partition()) |
| continue; |
| |
| auto ts = std::max(t1_.ts_start(), t2_.ts_start()); |
| res = t1_.StepUntil(ts); |
| if (PERFETTO_UNLIKELY(res.is_err())) |
| return res.err_code; |
| else if (PERFETTO_UNLIKELY(res.is_eof())) |
| continue; |
| |
| res = t2_.StepUntil(t1_.ts_start()); |
| if (PERFETTO_UNLIKELY(res.is_err())) |
| return res.err_code; |
| else if (PERFETTO_UNLIKELY(res.is_eof())) |
| continue; |
| |
| // If we're in the case where we have shadow slices on both tables, try |
| // and forward the earliest table and see what happens. IsOverlappingSpan() |
| // will double check that we have at least one non-real slice now. |
| // Note: if we don't do this, we end up in an infinite loop because all |
| // the code above will not change anything because these shadow slices will |
| // be overlapping. |
| if (!t1_.IsRealSlice() && !t2_.IsRealSlice()) { |
| PERFETTO_DCHECK(t1_.partition() == t2_.partition()); |
| |
| // If the table is not partitioned, partition() will return the partition |
| // the table was set to have by StepToPartition(). |
| auto t1_partition = |
| t1_.IsPartitioned() ? t1_.CursorPartition() : t1_.partition(); |
| auto t2_partition = |
| t2_.IsPartitioned() ? t2_.CursorPartition() : t2_.partition(); |
| |
| Query* stepped; |
| if (t1_partition == t2_partition) { |
| stepped = t1_.ts_end() <= t2_.ts_end() ? &t1_ : &t2_; |
| } else { |
| stepped = t1_partition <= t2_partition ? &t1_ : &t2_; |
| } |
| |
| res = stepped->Step(); |
| if (PERFETTO_UNLIKELY(res.is_err())) |
| return res.err_code; |
| else if (PERFETTO_UNLIKELY(res.is_eof())) |
| continue; |
| } |
| |
| if (IsOverlappingSpan()) |
| break; |
| } |
| next_stepped_ = t1_.ts_end() <= t2_.ts_end() ? &t1_ : &t2_; |
| |
| return SQLITE_OK; |
| } |
| |
| int SpanJoinOperatorTable::Cursor::Eof() { |
| return t1_.Eof() || t2_.Eof(); |
| } |
| |
| int SpanJoinOperatorTable::Cursor::Column(sqlite3_context* context, int N) { |
| PERFETTO_DCHECK(!t1_.Eof()); |
| PERFETTO_DCHECK(!t2_.Eof()); |
| |
| if (N == Column::kTimestamp) { |
| auto max_ts = std::max(t1_.ts_start(), t2_.ts_start()); |
| sqlite3_result_int64(context, static_cast<sqlite3_int64>(max_ts)); |
| } else if (N == Column::kDuration) { |
| auto max_start = std::max(t1_.ts_start(), t2_.ts_start()); |
| auto min_end = std::min(t1_.ts_end(), t2_.ts_end()); |
| PERFETTO_DCHECK(min_end > max_start); |
| auto dur = min_end - max_start; |
| sqlite3_result_int64(context, static_cast<sqlite3_int64>(dur)); |
| } else if (N == Column::kPartition && |
| table_->partitioning_ != PartitioningType::kNoPartitioning) { |
| PERFETTO_DCHECK(table_->partitioning_ == |
| PartitioningType::kMixedPartitioning || |
| t1_.partition() == t2_.partition()); |
| auto partition = t1_.IsPartitioned() ? t1_.partition() : t2_.partition(); |
| sqlite3_result_int64(context, static_cast<sqlite3_int64>(partition)); |
| } else { |
| size_t index = static_cast<size_t>(N); |
| const auto& locator = table_->global_index_to_column_locator_[index]; |
| if (locator.defn == t1_.definition()) |
| t1_.ReportSqliteResult(context, locator.col_index); |
| else |
| t2_.ReportSqliteResult(context, locator.col_index); |
| } |
| return SQLITE_OK; |
| } |
| |
| SpanJoinOperatorTable::Query::Query(SpanJoinOperatorTable* table, |
| const TableDefinition* definition, |
| sqlite3* db) |
| : defn_(definition), db_(db), table_(table) { |
| PERFETTO_DCHECK(!defn_->IsPartitioned() || |
| defn_->partition_idx() < defn_->columns().size()); |
| } |
| |
| SpanJoinOperatorTable::Query::~Query() = default; |
| |
| int SpanJoinOperatorTable::Query::Initialize(const QueryConstraints& qc, |
| sqlite3_value** argv) { |
| sql_query_ = CreateSqlQuery( |
| table_->ComputeSqlConstraintsForDefinition(*defn_, qc, argv)); |
| return PrepareRawStmt(); |
| } |
| |
| SpanJoinOperatorTable::Query::StepRet SpanJoinOperatorTable::Query::Step() { |
| PERFETTO_DCHECK(!Eof()); |
| sqlite3_stmt* stmt = stmt_.get(); |
| |
| // In this loop, we will try and find the slice to from the cursor. |
| // Terminology: "Shadow slices" are slices which fill in the gaps between real |
| // slices from the underlying cursor in each partition (if any). |
| // For queries which don't need "shadow slices", we simply return non-zero |
| // duration slices from the underlying cursor. |
| do { |
| if (mode_ == Mode::kShadowSlice) { |
| PERFETTO_DCHECK(defn_->emit_shadow_slices()); |
| |
| // If we're out of slices in the cursor, this shadow slice will be the |
| // final slice. |
| if (cursor_eof_) { |
| mode_ = Mode::kRealSlice; |
| return StepRet(StepRet::Code::kEof); |
| } |
| |
| // Look ahead to see if the cursor changes partition (if the cursor is |
| // partitioned). If so, then we need to fill the gap between the ts == 0 |
| // and the start of that slice. Otherwise after this slice, we have the |
| // real slice from the cursor. |
| if (!defn_->IsPartitioned() || partition_ == CursorPartition()) { |
| mode_ = Mode::kRealSlice; |
| ts_start_ = CursorTs(); |
| ts_end_ = ts_start_ + CursorDur(); |
| } else if (IsFullPartitionShadowSlice()) { |
| mode_ = Mode::kShadowSlice; |
| ts_start_ = 0; |
| ts_end_ = CursorTs(); |
| partition_ = CursorPartition(); |
| } else { |
| mode_ = Mode::kShadowSlice; |
| ts_start_ = 0; |
| ts_end_ = std::numeric_limits<int64_t>::max(); |
| } |
| continue; |
| } |
| |
| int res; |
| if (defn_->IsPartitioned()) { |
| auto partition_idx = static_cast<int>(defn_->partition_idx()); |
| // Fastforward through any rows with null partition keys. |
| int row_type; |
| do { |
| res = sqlite3_step(stmt); |
| row_type = sqlite3_column_type(stmt, partition_idx); |
| } while (res == SQLITE_ROW && row_type == SQLITE_NULL); |
| } else { |
| res = sqlite3_step(stmt); |
| } |
| |
| if (res == SQLITE_ROW) { |
| // After every row, there will be a shadow slice so emit that if we need |
| // to do so. Otherwise, just emit the underlying slice. |
| if (defn_->emit_shadow_slices()) { |
| mode_ = Mode::kShadowSlice; |
| ts_start_ = ts_end_; |
| ts_end_ = !defn_->IsPartitioned() || partition_ == CursorPartition() |
| ? CursorTs() |
| : std::numeric_limits<int64_t>::max(); |
| } else { |
| mode_ = Mode::kRealSlice; |
| ts_start_ = CursorTs(); |
| ts_end_ = ts_start_ + CursorDur(); |
| if (defn_->IsPartitioned()) |
| partition_ = CursorPartition(); |
| } |
| } else if (res == SQLITE_DONE) { |
| cursor_eof_ = true; |
| if (!defn_->emit_shadow_slices()) |
| return StepRet(StepRet::Code::kEof); |
| |
| // Close off the remainder of this partition with a shadow slice. |
| mode_ = Mode::kShadowSlice; |
| ts_start_ = ts_end_; |
| ts_end_ = std::numeric_limits<int64_t>::max(); |
| } else { |
| return StepRet(StepRet::Code::kError, res); |
| } |
| } while (ts_start_ == ts_end_); |
| |
| return StepRet(StepRet::Code::kRow); |
| } |
| |
| SpanJoinOperatorTable::Query::StepRet |
| SpanJoinOperatorTable::Query::StepToNextPartition() { |
| PERFETTO_DCHECK(defn_->IsPartitioned()); |
| PERFETTO_DCHECK(!Eof()); |
| |
| auto current_partition = partition_; |
| while (partition_ <= current_partition) { |
| auto res = Step(); |
| if (!res.is_row()) |
| return res; |
| } |
| return StepRet(StepRet::Code::kRow); |
| } |
| |
| SpanJoinOperatorTable::Query::StepRet |
| SpanJoinOperatorTable::Query::StepToPartition(int64_t target_partition) { |
| PERFETTO_DCHECK(partition_ <= target_partition); |
| if (defn_->IsPartitioned()) { |
| while (partition_ < target_partition) { |
| if (IsFullPartitionShadowSlice() && |
| target_partition < CursorPartition()) { |
| partition_ = target_partition; |
| return StepRet(StepRet::Code::kRow); |
| } |
| |
| auto res = Step(); |
| if (!res.is_row()) |
| return res; |
| } |
| } else if (/* !defn_->IsPartitioned() && */ partition_ < target_partition) { |
| int res = PrepareRawStmt(); |
| if (res != SQLITE_OK) |
| return StepRet(StepRet::Code::kError, res); |
| partition_ = target_partition; |
| } |
| return StepRet(StepRet::Code::kRow); |
| } |
| |
| SpanJoinOperatorTable::Query::StepRet SpanJoinOperatorTable::Query::StepUntil( |
| int64_t timestamp) { |
| PERFETTO_DCHECK(!Eof()); |
| auto partition = partition_; |
| while (partition_ == partition && ts_end_ <= timestamp) { |
| auto res = Step(); |
| if (!res.is_row()) |
| return res; |
| } |
| return StepRet(StepRet::Code::kRow); |
| } |
| |
| std::string SpanJoinOperatorTable::Query::CreateSqlQuery( |
| const std::vector<std::string>& cs) const { |
| std::vector<std::string> col_names; |
| for (const Table::Column& c : defn_->columns()) { |
| col_names.push_back("`" + c.name() + "`"); |
| } |
| |
| std::string sql = "SELECT " + base::Join(col_names, ", "); |
| sql += " FROM " + defn_->name(); |
| if (!cs.empty()) { |
| sql += " WHERE " + base::Join(cs, " AND "); |
| } |
| sql += " ORDER BY "; |
| sql += defn_->IsPartitioned() |
| ? base::Join({"`" + defn_->partition_col() + "`", "ts"}, ", ") |
| : "ts"; |
| sql += ";"; |
| PERFETTO_DLOG("%s", sql.c_str()); |
| return sql; |
| } |
| |
| int SpanJoinOperatorTable::Query::PrepareRawStmt() { |
| sqlite3_stmt* stmt = nullptr; |
| int err = |
| sqlite3_prepare_v2(db_, sql_query_.c_str(), |
| static_cast<int>(sql_query_.size()), &stmt, nullptr); |
| stmt_.reset(stmt); |
| |
| ts_start_ = 0; |
| ts_end_ = 0; |
| partition_ = std::numeric_limits<int64_t>::lowest(); |
| cursor_eof_ = false; |
| mode_ = Mode::kRealSlice; |
| |
| return err; |
| } |
| |
| void SpanJoinOperatorTable::Query::ReportSqliteResult(sqlite3_context* context, |
| size_t index) { |
| if (mode_ != Mode::kRealSlice) { |
| sqlite3_result_null(context); |
| return; |
| } |
| |
| sqlite3_stmt* stmt = stmt_.get(); |
| int idx = static_cast<int>(index); |
| switch (sqlite3_column_type(stmt, idx)) { |
| case SQLITE_INTEGER: |
| sqlite3_result_int64(context, sqlite3_column_int64(stmt, idx)); |
| break; |
| case SQLITE_FLOAT: |
| sqlite3_result_double(context, sqlite3_column_double(stmt, idx)); |
| break; |
| case SQLITE_TEXT: { |
| // TODO(lalitm): note for future optimizations: if we knew the addresses |
| // of the string intern pool, we could check if the string returned here |
| // comes from the pool, and pass it as non-transient. |
| const auto kSqliteTransient = |
| reinterpret_cast<sqlite3_destructor_type>(-1); |
| auto ptr = reinterpret_cast<const char*>(sqlite3_column_text(stmt, idx)); |
| sqlite3_result_text(context, ptr, -1, kSqliteTransient); |
| break; |
| } |
| } |
| } |
| |
| SpanJoinOperatorTable::TableDefinition::TableDefinition( |
| std::string name, |
| std::string partition_col, |
| std::vector<Table::Column> cols, |
| bool emit_shadow_slices, |
| uint32_t ts_idx, |
| uint32_t dur_idx, |
| uint32_t partition_idx) |
| : name_(std::move(name)), |
| partition_col_(std::move(partition_col)), |
| cols_(std::move(cols)), |
| emit_shadow_slices_(emit_shadow_slices), |
| ts_idx_(ts_idx), |
| dur_idx_(dur_idx), |
| partition_idx_(partition_idx) {} |
| |
| base::Optional<SpanJoinOperatorTable::TableDescriptor> |
| SpanJoinOperatorTable::TableDescriptor::Parse( |
| const std::string& raw_descriptor) { |
| // Descriptors have one of the following forms: |
| // table_name [PARTITIONED column_name] |
| |
| // Find the table name. |
| base::StringSplitter splitter(raw_descriptor, ' '); |
| if (!splitter.Next()) |
| return base::nullopt; |
| |
| TableDescriptor descriptor; |
| descriptor.name = splitter.cur_token(); |
| if (!splitter.Next()) |
| return std::move(descriptor); |
| |
| if (strcasecmp(splitter.cur_token(), "PARTITIONED") != 0) { |
| PERFETTO_ELOG("Invalid SPAN_JOIN token %s", splitter.cur_token()); |
| return base::nullopt; |
| } |
| if (!splitter.Next()) { |
| PERFETTO_ELOG("Missing partitioning column"); |
| return base::nullopt; |
| } |
| |
| descriptor.partition_col = splitter.cur_token(); |
| return std::move(descriptor); |
| } |
| |
| } // namespace trace_processor |
| } // namespace perfetto |
