• Created by Unknown User (morningman), last updated on Dec 27, 2022 8 minute read

Status

Current state: Released

Discussion thread: 

JIRA or Github Issue: 

Released: 1.2.0

Google Doc: <If the design in question is unclear or needs to be discussed and reviewed, a Google Doc can be used first to facilitate comments from others.>

Motivation

Doris currently supports the external table. This feature is through the mapping of metadata, so that Doris can access external data sources. We take the iceberg data source as an example, which has two metadata mapping methods:

  1. Create table directly
    In this way, the one-to-one mapping relationship between the table in the Doris metadata and the table in the Iceberg metadata is completed directly by establishing an external table.
  2. Create database with external datasource property
    Set the url of iceberg metadata service(such as hive metastore) in the database property. This is equivalent to establishing a mapping relationship between database in Doris metadata and a database (or schema) in iceberg.
    After that, the Doris will automatically synchronize all tables under this database in background thread, and automatically complete the one-to-one mapping relationship of metadata of all tables.

We can see that the above two methods complete the metadata mapping at the database level and table level respectively. If the iceberg cluster has multiple databases, at least we need to manually establish multiple database-level mapping relationships. This is inconvenient to use in some scenarios.

In many scenarios, the user hopes to implement the "data source" level mapping, that is, through the "data source" level mapping relationship, Doris can directly access all data under the corresponding data source, including database and table. In this way, the user can rely on Doris and have unified access to multiple "data sources".

Therefore, the motivation of this feature is to add a new metadata level called "datasource" to Doris to support connecting and accessing to other data sources.

Related Research

Just like Presto, there is Catalog→ Database→Table hierarchical relationships to manage metadata.

In Doris, I would like to introduce a new metadata called DataSource, which is exactly same as Catalog in Presto.

1-DataSource

The current metadata hierarchy of Doris is Database->Table->Column.
We add a new level: DataSource->Database->Table→Column.

A DataSource represents a data source, such as an Iceberg cluster, a Hive cluster, an ES cluster, or another Doris cluster.

We can define data sources into the following two categories:

  1. Internal DataSource
    Each Doris cluster will have an Internal DataSource by default, which manages all databases under the current cluster. When the upgrade Doris from old version, this Datasource will be added by default.
  2. External Datasource
    All data sources that are not self-manager by Doris will be referred to as External Datasource. Such as hive, iceberg, hudi, es, odbc, etc. Users can create several External DataSources.
    External DataSource is a base class, we can provide different subclass implementations according to different data sources, such as:
    1. Hive Metastore: Used to access hive metastore compatible data sources.
    2. ES: Support ES data source
    3. ODBC: Supports accessing external data sources through odbc.
    4. Doris: Other Doris cluster

2-How to manager DataSource

Here we discuss about how to manager datasources, for example, how to add a new databases.

  1. Built-in
    Like the various external data sources currently implemented by Doris, each external data source requires separate development and adaptation, and user need to upgrade Doris to new version to access new data source. This method is expensive, and it is difficult to solve diversity problems such as different versions of hive.
  2. SPI
    Similar to Presto's connector, the interface is defined, and the user completes the specific implementation, and then accesses it through dynamic class loader. This approach can solve the problem of data source diversity. But here are a few issues to consider:
    1. What are the needs of most users: For most users, there are only a few data sources that may be concerned, such as hive, iceberg, hudi, odbc, ES. If we have built-in support for some or all of them, then there is nothing need to be done by user.
    2. We can use SPI on FE side. But on BE side, use dynamic library will have many compatibility problems. Therefore, only the metadata part can be customized by users, but the data access on BE side cannot be customized.
      Therefore, in the data access layer, we need to unify the access methods to a limited number of ways as much as possible to ensure that the built-in data access layer (BE side) can support most of the data sources, for example:
          1. HDFS
          2. S3
          3. ODBC
      For more access methods, we can consider another form, that is to deploy an Http Connector Service on the user's data source side. The specificity of data source access is masked through this HTTP Service. For Doris, it boils down to HTTP access. This method can bring more flexible scalability and reduce the intrusion to the Doris cluster. For example, it can handle network connectivity issues, authorization verification, data caching and even data filtering. Of course, this method brings extra maintenance costs to users, and also has high requirements on the availability and performance of the http connection service itself.

To sum up, we will combine the above methods to carry out the first stage design: Built-in data source + SPI interface

  1. Built-in data sources meet most usage scenarios, and SPI meets diverse needs
  2. Unify the interface definition of built-in data source and SPI to ensure the unity of code logic

Also, the user-side http connection service can be extended based on this interface in next stage.

3-Metadata Cache

In the previous implementation, all external tables were persisted in Doris' metadata.

In the new implementation, these metadata are no longer persistent, but the metadata of the external data source is obtained through real-time access to the datasource.

Of course, this may have performance issues, so we need to design a metadata caching strategy to improve efficiency.

4-Qualifier

The qualified names in SQL is divided into three categories:

  1. Column: Column name. Fully qualified name: db.table.column
  2. Table: Table name. Fully qualified name: db.table
  3. UDF: user define function, the UDF belongs to a certain db, so the fully qualified name is db.function()

After adding the datasource level, all qualified names need to add a extra level.
For Tables and UDFs, adding one level does not create semantic ambiguity:

  • A.B.C: ds.db.tbl/func
  • B.C: db.tbl/func
  • C: tbl/func

For Column, because we will support nested types in the future, and fieds of nested types are still connected using DOT. Then there may be ambiguity here, give an example.
Column col: Struct type Col <M1, M2>
Then it might be written as follows:

  • ds.db.tbl.col.m1
  • db.tbl.col
  • tbl.col.m1

You can see that the second and third formats are the same, but the content user want to express is different. So we need a priority-based qualified name resolution rule.

3 parts or less: Match from right to left. This approach ensures forward compatibility. After the datasource level is added, the user does not need to change the original SQL statement.

Example:

  • A.B.C: (d.t.c)->(t.c.f1)→(c.f1.f2)
  • A.B: (t.c)→(c.f1)
  • C: (c)

4 parts or more: Match from left to right.

Example:

  • A.B.C.D: (c.f1.f2.f3)->(t.c.f1.f2)->(d.t.c.f1)→(ds.d.t.c)
  • A.B.C.D.E: (c.f1.f2.f3.f4)->(t.c.f1.f2.f3)->(d.t.c.f1.f2)→(ds.d.t.c.f1)

The modification of the qualified name should only affect the SQL query, because we want to support data federation query between multiple datasources.

But for other functions, we can temporarily not support cross-datasource operations, such as DDL and DML.

In this way, in order not to affect the original syntax, we need to add an additional syntax similar to `use db`, first set the datasource to which the current session belongs, and then all non-query operations are completed within this datasource:

`switch datasource`

5-Privilege

The current Doris privilege management refers to the hierarchical design of mysql, with only has 3 level(global, database and table). Here we need to add a new level for data source.

6-Forward Compatibility

This solution needs to ensure that the original cluster can be upgraded normally.
First, after the old cluster is upgraded, all existing databases in the original cluster will be included under the default internal datasource. This ensures that the original function is completely unchanged. New datasources are created through the new create datasource syntax.

In addition, the previous external table functions are all retained, which can be understood as the external tables created before belong to the Internal Datasource.

7-Information_schema

information_schema remains unique across the cluster. Because the datasource level is already included in the information_schema, multiple datasources can be stored through the information_schema.

Detailed Design

1-Create datasource

create [external] catalog my_datasource properties("k" = "v", ...);
drop catalog my_datasource;
show catalogs;
modify catalog my_datasource set("k" = "v", ...);

2-Privilege Management

# No additional syntax will be added. If you need to authorize privilege under other datasource, you need to switch ds first.
# During initialization, you can use the admin account to switch to the new ds to create the permissions of the initial account.
grant all on db.tbl to user@%;


3-Basic data structure

                                                                +---------------------+
                                                           +--->|HMSExternalDataSource|
                                                           |    +---------------------+
                               +-----------------------+   |    +---------------------+
                               |ExternalDataSource     |   +--->|EsExternalDataSource |
                               |                       +---+    +---------------------+
                         +----->    listDatabaseNames()|   |    +---------------------++
                         |     |    tableExists()      |   +--->|OdbcExternalDataSource|
+-------------------+    |     |    getTableSchema()   |   |    +----------------------+
| DataSourceIf      |    |     |    ...                |   |    +----------------------+
|                   +----+     +-----------------------+   +--->|HttpExternalDataSource|
|    getDatabase()  |                                           +----------------------+
|    listDatabases()+----+     +-----------------------+
|    ...            |    |     |InternalDataSource     |
+-------------------+    |     |                       |
                         +----->    self-managed meta  |
                               |                       |
                               +-----------------------+

+-------------------+          +-----------------------+
| DatabaseIf        +---------->ExternalDatabase       |
|                   |          +-----------------------+
|    getTable()     |
|    listTable()    |          +-----------------------+
|    ...            +---------->Database               |
+-------------------+          +-----------------------+


+-------------------+          +-----------------------+
| TableIf           +---------->ExternalTable          |
|                   |          +-----------------------+         +----------+
|    getColumn()    |                                      +---->|OlapTable |
|    getSchema()    |          +-----------------------+   |     +----------+
|    ...            +---------->Table                  +---+     +----------+
+-------------------+          +-----------------------+   +---->|OdbcTable |
                                                                 +----------+


Interfaces:

  • DataSourceIf: provide getDatabase(), listDatabases(), etc.
  • DatabaseIf: provide getTable(), listTables(), etc.
  • TableIf: provide getColumn(), getSchema(), etc.


Classes:

  • InternalDataSource implements DataSourceIf: contains all self-managed metadata.
  • ExternalDataSource implements DataSourceIf: contains all not-self-managed metadata which should be got from external data source.
    • HMSExternalDataSource: hive metastore compatible data sources.
    • EsExternalDataSource: elasticsearch
    • OdbcExternalDataSource: odbc compatible data sources
    • HttpExternalDataSource: for http api
    • ...


  • ExternalDatabase implements DatabaseIf: database got from external data source.
  • Database implements DatabaseIf: the self-managed databases.


  • ExternalTable implements TableIf: table got from external data source.
  • Table implements TableIf: the self-managed tables
    • OlapTable
    • MysqlTable
    • OdbcTable
    • ...


  • ExternalScanNode: for external table scan task.


  • MetaObjCache: To cache and manage all external meta objects.

Scheduling

  1. Creation of DataSource (In review)
    1. create/drop/alter/show/switch syntax support
    2. Persistence of DataSource
    3. https://github.com/apache/incubator-doris/pull/10033
  2. InternalDataSource Support (Already merged)
    1. The change of the original interface ensures that the previous use logic remains unchanged.
    2. https://github.com/apache/incubator-doris/pull/9953
    3. https://github.com/apache/incubator-doris/pull/10044
  3. HMSExternalDataSource Support (Under development, 6.17)
    1. Ability to synchronize HiveMetaStore data sources
    2. Able to show databases, show tables, desc table, information schema
    3. Can be converted to the original Hive/HudiScanNode for data query
    4. https://github.com/apache/incubator-doris/pull/10028
  4. FE side ExternalFileScanNode development (Under development, 6.17)
    1. Unify hive/iceberg/hudi scan node
    2. Support partition prunning and other functions
    3. https://github.com/apache/incubator-doris/pull/9973
  5. BE side ExternalFileScanNode development (6.24)
    1. Replace the BrokerScanNode
    2. Only supported in the vectorized engine, with the vectorized parquet/orc reader
  6. DataSource permission management (Under development, 6.17)
    1. Support DataSource-level permission management
  7. SQL syntax support (6.24)
    1. SQL syntax supports three-level qualified names
    2. Support query across DataSource
  8. Parquet predicate pushdown (Designing, 6.24)
    1. Support column pruning and predicate filtering for parquet files
  9. Test and Bug Rush (6.30)
    1. [3-7] test of part of the work
    2. Functionally the same as before

At the end of June, support access to hive, iceberg, hudi through the new framework, and the performance is better than before.

==================================================

  1. Read optimization of ORC/Text file format (ByteDance)
  2. ESExternalDataSource support
  3. ODBCExternalDataSource support
  4. Support for appearance statistics
  5. Collection and display interface of statistics
  6. Statistics framework that needs to be combined with the new optimizer
  7. Meta Object Cache




  • No labels