frontend/instance/
copy_in.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use api::v1::RowInsertRequests;
16use async_trait::async_trait;
17use auth::{
18    PermissionChecker, PermissionCheckerRef, PermissionReq, PermissionTableTarget,
19    PermissionTableTargets,
20};
21use common_error::ext::BoxedError;
22use common_query::Output;
23use query::query_engine::options::{QueryOptions, validate_catalog_and_schema};
24use servers::error::{AuthSnafu, ExecuteGrpcRequestSnafu, ExecuteQuerySnafu, Result};
25use servers::query_handler::CopyInHandler;
26use session::context::QueryContextRef;
27use snafu::ResultExt;
28use table::metadata::TableInfoRef;
29
30use crate::instance::Instance;
31
32#[async_trait]
33impl CopyInHandler for Instance {
34    async fn copy_in_table(
35        &self,
36        catalog: &str,
37        schema: &str,
38        table: &str,
39        query_ctx: QueryContextRef,
40    ) -> Result<Option<TableInfoRef>> {
41        // Mirror SQL INSERT's catalog restriction (see `check_permission`):
42        // when cross-catalog queries are disallowed, a COPY may only
43        // target the current catalog. This must run against the ORIGINAL
44        // query context — `begin_copy_in` switches the context to the
45        // target catalog afterwards, so nothing downstream would catch
46        // it.
47        let disallow_cross_catalog_query = self
48            .plugins
49            .get::<QueryOptions>()
50            .map(|opts| opts.disallow_cross_catalog_query)
51            .unwrap_or_default();
52        if disallow_cross_catalog_query {
53            validate_catalog_and_schema(catalog, schema, &query_ctx)
54                .map_err(BoxedError::new)
55                .context(ExecuteQuerySnafu)?;
56        }
57
58        // A COPY writes rows like a SQL INSERT; check the write permission
59        // before the client starts streaming data.
60        self.plugins
61            .get::<PermissionCheckerRef>()
62            .as_ref()
63            .check_permission(
64                query_ctx.current_user(),
65                PermissionReq::BulkInsert {
66                    catalog,
67                    schema,
68                    table,
69                },
70            )
71            .context(AuthSnafu)?;
72
73        let table_info = self
74            .catalog_manager()
75            .table(catalog, schema, table, None)
76            .await
77            .map(|table| table.map(|table| table.table_info()))
78            .map_err(servers::error::Error::from)?;
79        let target = PermissionTableTarget::new(catalog, schema, table).with_physical_metric_table(
80            table_info
81                .as_ref()
82                .is_some_and(|info| info.is_physical_table()),
83        );
84        self.check_table_permission(
85            &query_ctx,
86            PermissionReq::BulkInsert {
87                catalog,
88                schema,
89                table,
90            },
91            PermissionTableTargets::resolved(vec![target]),
92        )
93        .context(AuthSnafu)?;
94        Ok(table_info)
95    }
96
97    async fn copy_in_insert(
98        &self,
99        requests: RowInsertRequests,
100        query_ctx: QueryContextRef,
101    ) -> Result<Output> {
102        self.handle_row_inserts(requests, query_ctx, false, false)
103            .await
104            .map_err(BoxedError::new)
105            .context(ExecuteGrpcRequestSnafu)
106    }
107}
frontend/instance/copy_in.rs

frontend/instance/
copy_in.rs
