glaredb / glaredb Goto Github PK

In particular, the coretypes crate has been replaced with the lemur crate.

lemur and sqlengine are the two core crates we need to make sure to keep. diststore has an accord module that might be useful to use/revamp in the future.

Currently only the resulting dataframe from executing a query is returned.

I don't know how we want to return the columns yet, but all the information required is in the the Scope struct.

Testing please ignore.

This might result in a large diff, so splitting into multiple PRs may be preferred.

Will require and, and_scalar, or and or_scalar implementations in the column compute module.

E.g.

pub fn and(left: &Column, right: &Column) -> Result<Column> {
    let left = left
        .try_downcast_bool()
        .ok_or_else(|| internal!("left not a bool column"))?;
    let right = right
        .try_downcast_bool()
        .ok_or_else(|| internal!("right not a bool column"))?;
    let out = arrow2::compute::boolean::and(left.0, right.0);
    Ok(out.into())
}

Hash aggregate, should support grouping sets (since this leads to supporting rollups).

For each scalar expression, we should to constant fold as much as possible. This pretty much entails walking scalar expressions and folding subtrees that don't contain a Column or Aggregate into a Constant.

A method should be added to PlanExpr for this which takes self and produces a new PlanExpr. And then during plan rewriting, we can just call that method for each expression we encounter.

• When do we ensure that all the required tables for the catalog exist?
  • Note that the first iteration will probably have everything checked/created on startup. Eventually we would want to add global lock or pessimistic transaction to ensure only one node is creating the catalogs on startup and that all other block/wait on it.
• What catalogs do we need?
  • https://www.cockroachlabs.com/docs/stable/system-catalogs.html
  • https://www.postgresql.org/docs/current/catalogs.html
  • Might be enough to have a subset INFORMATION_SCHEMA.
• How will catalogs be defined in the code?

The primary goal here is to get a database catalog up and running to allow for simple queries (create, insert, select, query). Adding index information should come later.

A previous version of the sqlengine crate had a very minimal proof-of-concept catalog.

Implement a subset of the Postgres wire protocol (here) to allow for execution of simple queries.

There's currently no concept of users/authentication, so when implementing AuthenticationCleartextPassword, it's sufficient to use a hardcoded username and an empty password.

Inspirations:

• https://github.com/MaterializeInc/materialize/tree/main/src/pgwire (very complete)
• https://github.com/chotchki/feophant/tree/main/src/codec

See also msql-srv for how this could be structured in a way to allow stubbing out the database itself. This isn't necessary to do, but just might make testing easier.

See #38 (comment)

The currently filter pushdown rewriter only pushes a filtering expression down if the child node is a ScanSource node. We should be doing similar with joins.

There's two things of note here; we should be able to push down predicates in actual Filter nodes, as well as predicates in the Join node as well.

For example, this query produces the following plan.

select * from t1 inner join t2 on t1.b = t2.b where t1.a > 5

plan: Read(
    Project(
        Project {
            columns: [
                Column(
                    0,
                ),
                Column(
                    1,
                ),
                Column(
                    2,
                ),
                Column(
                    3,
                ),
            ],
            input: Filter(
                Filter {
                    predicate: Binary {
                        op: Gt,
                        left: Column(
                            0,
                        ),
                        right: Constant(
                            Int8(
                                Some(
                                    5,
                                ),
                            ),
                        ),
                    },
                    input: Join(
                        Join {
                            left: ScanSource(
                                ScanSource {
                                    table: TableReference {
                                        catalog: "catalog",
                                        schema: "schema",
                                        table: "t1",
                                    },
                                    filter: None,
                                    schema: Schema {
                                        types: [
                                            Int8,
                                            Int8,
                                        ],
                                    },
                                },
                            ),
                            right: ScanSource(
                                ScanSource {
                                    table: TableReference {
                                        catalog: "catalog",
                                        schema: "schema",
                                        table: "t2",
                                    },
                                    filter: None,
                                    schema: Schema {
                                        types: [
                                            Int8,
                                            Int8,
                                        ],
                                    },
                                },
                            ),
                            join_type: Inner,
                            on: Binary {
                                op: Eq,
                                left: Column(
                                    1,
                                ),
                                right: Column(
                                    2,
                                ),
                            },
                        },
                    ),
                },
            ),
        },
    ),
)

Note that the filter is above the join, and no filters are provided to any of the scans. We (I think) have enough context in the plan to be able to push the above filters and predicates into their respective scans.

This will require extracting sub expressions for more complex expressions that reference both sides of the join. For example, the filtering expresion t1.a > 5 and t2.b > 10 would need to be split to pass the relevant sub expressions in the source table scans.

The primary use case will be for capping memory usage during query execution. Right now, we can be very coarse with how we determine how much memory an executor is using.

This will also be used to allow for disk spillage once we run out memory. Disk spillage won't be too fancy, and we'll make use of arrow2's io utilities for writing out to files.

This will need to be done after the majority of the query executors are implemented so that we have a good understanding of what we need from the execution context. In the meantime, we should note in the code where we should be spilling.

This will require a minimal design doc prior to implementation.

There's currently no validation at that level, and we're relying on the storage engine to return an error in such cases.

This will require that data source supports index scan (and we'll need to come up with a decent api for it).

(IsNull and IsNotNull)

This will require adding traits/implementations to the repr/compute module in lemur.

https://github.com/hyperium/tonic

See https://github.com/erikgrinaker/toydb/blob/master/src/sql/engine/raft.rs for an example of how this might look.

Related to the arrow2 refactor. Blocked on having an appropriate "mutating" interface.

This will give use the opportunity to think through how we want to structure our data is RocksDB, both in terms of indexes as well as transactional semantics.

Currently only support "primary" indexes (see https://github.com/GlareDB/glaredb/blob/main/crates/storageengine/src/repr.rs).

We'll need to design a layout to support secondary indexes, as well as transaction semantics (e.g. write intents).

The approximate api should look something like the following:

async fn scan_values_equal(&self, table: &RelationKey, values: &[(usize, Value)]) -> Result<Option<DataFrameStream>>

values here represents a list of tuples with a column index and value expected to be found for that column.

This this can be added as a default implementation on the trait, since it can make use of the existing scan method.

The current RocksStore has some stuff related to transactions. It currently uses an atomic u64 for generating transaction ids. These ids are not currently referenced when inserting or retrieving data from RocksDB.

This issue encompasses:

• Introducing key representations that include transaction timestamps for MVCC.
• Introducing/modifying the RocksTx type.
  • Holds the transaction timestamp.
  • Implements the new lemur execution interfaces.
• Introduces "write intents" to allow for interactive transactions.

A RocksTx should be able to be instantiated with an arbitrary timestamp. Timestamps will be generated from a hybrid logical clock. This can be used both for running locally as well as running in a cluster. When running locally, the "node id" portion of the timestamp can just be 0.

When converting the HLC to bytes, order must be maintained (such that byte representations of HLCs can be compared directly).

This will require a minimal design doc (particularly for key/value layouts and how those will fit in with secondary indexes).

Followup from #22

This will allow us to make use of the projection fields on table scan nodes added in the above issue.

This will inspect Project nodes, extract column indices, place those in the table scan nodes.

As example, let's say we had table t1 with the following columns:

A plan for select b, b + 100 from t1 would have a Projection node with the expressions Column(1), BinaryOp{op: Add, left: Column(1), right: Constant(100)}. Pushing this down is relatively straightforward, we go through each expression and pull out every column index that we find, deduplicate, then add those indices to the projection field on table scan node. In this case, our projection field would look like Some(vec![1]).

Once we have the indeces we want to project, it's important that we rewrite the original expression to reference the appropriate columns after projection. In this case, our expressions would be Column(0), BinaryOp{op: Add, left: Column(0), right: Constant(100)} since the table scan will only be returning a dataframe with a single column (which happens to be column b).

Note that joins and aggregates will add complexity. The above example is the one of the simpler cases.

In some cases (particularly on the storage side), we don't have a full "data frame" available, and instead we're working with sets of rows. In those cases, it's inefficient having to transform a row into data frame before being able execute an expression on it.

This may require some changes to the traits related to execution, but I'm unsure of how much.

This issue should be completed in conjunction with building out the storage engine, as that's the primary area where we'll want to execute on individual rows.

get(&self, idx) -> Option<Option<...>>

Should take into account the validity at that index.

Miri is quite slow, so we need to be selective about which crates to test.

Also, miri requires recompiling everything, so we'll have to figure out the caching situation there. I don't want it overwriting the build cache that we're using for testing.

Related to the arrow2 refactor.

Executors:

• Project
• Values
• Aggregate¹
• Sort
• Cross Join
• Filter
• Hash Join¹
• Index Scan
• Table Scan
• Merge Join¹

https://www.sqlite.org/sqllogictest/doc/trunk/about.wiki

Tests if the output of a sql query is correct. Does not attempt say anything about performance.

DuckDB uses this exensively for their testing (see test_join.test for an example).

Probably GCR.

This image can be used for an mvp cloud where we spin up containers on k8s.

Planning currently just plans inner and cross joins.

We should properly plan left, right, and full outer joins as well.

I'm currently unsure what it would take to represent these joins as a CrossJoin followed by a Filter.

Ideally we're as up to date as possible for the major crates we're building around (datafusion).

E.g. executing select 1 should return a data frame with one row containing the value 1. It currently returns zero rows.

Queries often times only need a subset of columns, so we should be able to pass in Option<&[usize]> (or Option<Vec<usize>>) to select which columns we want to return. Note that this is using column indices and not a ScalarExpression since we want to try to return a minimal amount of data (since this will eventually be going over the network). An expression might produce more data than necessary, and that extra work is better off being done on the "compute" side of things.

E.g. if we were to pass in an expression for the query select a, a * 10 from t, we would have to return a data frame containing both the values for a and a * 10. Whereas passing in indices means the storage source is only returning a, and a * 10 can be computed later.

This'll required adding an additional field to the scan nodes in lemur's RelationExpr and sqlengine's ReadPlan. During initial planning, this field will be initialized to None.

Should be enough to call ensure_system_tables for now.

Ideally I should be able to create a test table in the client repl:

~/Code/github.com/glaredb/glaredb [2] $ cargo run --bin glaredb -- client localhost:6543
    Finished dev [unoptimized + debuginfo] target(s) in 0.04s
     Running `target/debug/glaredb client 'localhost:6543'`
connected to GlareDB
glaredb> create table test (a int)
error: unexpected response: Error("missing table system.gl_internal.columns")
glaredb>

Note: In the future, we might want to look into more synchronization to ensure only one node in the cluster is creating these tables, either with an explicit "bootstrap" mode, or by relying on transactional semantics.

Currently there's only cross join. When lowering to lemur, all joins are converted to a cross join followed by a filter node. This is obviously not efficient.

The current cross join implementation keeps all dataframes in memory, and we'll want to do the same for these other joins as well for the time being.

Nested loop would probably be the most straightforward to implement (the existing cross join implementation is pretty much a nest loop join with no predicate). Merge joins should also be somewhat straightforward.

Hash joins will require a bit more design work. We'll also need to be able to detect when a join predicate is an equality predicate between two tables.

Ideally this is a separate step from the "test" step so that it's easy to see at a glance if tests pass on PRs or not.

Implement the RaftNetwork and RaftNetworkFactory traits for openraft.

Openraft has a lot of types that need to be implemented, but for networking, these are relatively straightforward. They provide a BasicNode implementation that we can use for the Node type in the type config. NodeId can just be a type alias to u64.

Note that we want to use the main branch of openraft (pinned to specific commit in Cargo.toml). There's some api changes between what's in main and the docs on docs.rs. I would recommend pulling down openraft and running cargo doc locally.

This should have the same transactional semantics as inserting data.

We might be able to get some inspiration from existing solutions:

• https://rockset.com/blog/how-rocksets-converged-index-powers-real-time-analytics/
• https://github.com/cockroachdb/cockroach/blob/master/docs/tech-notes/encoding.md#indexes-secondary-indexes

This will require a minimal design doc.

I like the idea of nix, and we already a few nix files in the repo. Unfortunately, nix can be hard to pick up which might cause a bit a chafing as new devs are onboarded.

I'm proposing that we have section in the README (or a separate CONTRIBUTING doc) detailing how to use nix (e.g. opening a dev shell, creating a docker image, etc). This should be aimed at someone who hasn't used nixed before, so we need to make sure things like needing to enable flakes and the new nix3 command are accounted for.

The current implementation is here:
https://github.com/GlareDB/glaredb/blob/main/crates/lemur/src/execute/stream/read.rs#L258-L275

We'll need to implement it for the arrow2 refactor:
https://github.com/GlareDB/glaredb/blob/arrow2/crates/lemur/src/arrow/queryexec/filter.rs

Everything will be working on Chunk, and the Chunk type has an appropriate filter method. The flow is as follows:

• For each chunk, execute the scalar expression against that chunk.
• Get the column result from the expression result, and down cast to a BoolColumn.
• Call filter with the downcasted BoolColumn.
• Return the result.

https://docs.rs/futures/latest/futures/prelude/stream/trait.StreamExt.html#method.map will likely be the method of choice to use on the input stream.