FrankChen021 commented on code in PR #19107: URL: https://github.com/apache/druid/pull/19107#discussion_r3208936475
########## extensions-contrib/openlineage-emitter/src/main/java/org/apache/druid/extensions/openlineage/OpenLineageRequestLogger.java: ########## @@ -0,0 +1,626 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + */ + +package org.apache.druid.extensions.openlineage; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.apache.calcite.avatica.util.Casing; +import org.apache.calcite.avatica.util.Quoting; +import org.apache.calcite.sql.SqlBasicCall; +import org.apache.calcite.sql.SqlCall; +import org.apache.calcite.sql.SqlIdentifier; +import org.apache.calcite.sql.SqlInsert; +import org.apache.calcite.sql.SqlJoin; +import org.apache.calcite.sql.SqlKind; +import org.apache.calcite.sql.SqlNode; +import org.apache.calcite.sql.SqlNodeList; +import org.apache.calcite.sql.SqlSelect; +import org.apache.calcite.sql.SqlWith; +import org.apache.calcite.sql.SqlWithItem; +import org.apache.calcite.sql.parser.SqlParseException; +import org.apache.calcite.sql.parser.SqlParser; +import org.apache.druid.java.util.common.concurrent.Execs; +import org.apache.druid.java.util.common.lifecycle.LifecycleStart; +import org.apache.druid.java.util.common.lifecycle.LifecycleStop; +import org.apache.druid.java.util.common.logger.Logger; +import org.apache.druid.query.BaseQuery; +import org.apache.druid.server.RequestLogLine; +import org.apache.druid.server.log.RequestLogger; +import org.apache.druid.utils.CloseableUtils; +import org.apache.http.client.HttpClient; +import org.apache.http.client.methods.HttpPost; +import org.apache.http.entity.ContentType; +import org.apache.http.entity.StringEntity; +import org.apache.http.util.EntityUtils; + +import javax.annotation.Nullable; +import java.io.Closeable; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.RejectedExecutionHandler; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * OpenLineage RunEvents for completed Druid queries. + */ +public class OpenLineageRequestLogger implements RequestLogger +{ + private static final Logger log = new Logger(OpenLineageRequestLogger.class); + + private static final String PRODUCER = + "https://github.com/apache/druid/extensions-contrib/openlineage-emitter";; + private static final String SCHEMA_URL = + "https://openlineage.io/spec/2-0-2/OpenLineage.json";; + private static final String ENGINE_FACET_SCHEMA_URL = + "https://openlineage.io/spec/facets/1-1-1/ProcessingEngineRunFacet.json";; + private static final String ERROR_FACET_SCHEMA_URL = + "https://openlineage.io/spec/facets/1-0-0/ErrorMessageRunFacet.json";; + private static final String JOB_TYPE_FACET_SCHEMA_URL = + "https://openlineage.io/spec/facets/2-0-2/JobTypeJobFacet.json";; + private static final String SQL_FACET_SCHEMA_URL = + "https://openlineage.io/spec/facets/1-0-1/SQLJobFacet.json";; + + // Standard Calcite parser (not Druid's custom parser): sufficient for table name extraction. + // Druid-specific syntax (REPLACE, EXTERN, etc.) falls through to the SqlParseException handler. + private static final SqlParser.Config SQL_PARSER_CONFIG = SqlParser + .config() + .withCaseSensitive(true) + .withUnquotedCasing(Casing.UNCHANGED) + .withQuotedCasing(Casing.UNCHANGED) + .withQuoting(Quoting.DOUBLE_QUOTE); + + // Matches INSERT INTO or REPLACE INTO when the standard Calcite parser fails (e.g. EXTERN, + // PARTITIONED BY). Captures the output table name; handles quoted and unquoted identifiers. + private static final Pattern WRITE_INTO_PATTERN = + Pattern.compile("(?i)^\\s*(?:INSERT|REPLACE)\\s+INTO\\s+\"?([^\"\\s]+)\"?"); + // Extracts the SELECT subquery from INSERT INTO or REPLACE INTO for input lineage. + // Stops before PARTITIONED BY / CLUSTERED BY which are Druid-specific clauses. + private static final Pattern SELECT_FROM_WRITE_PATTERN = + Pattern.compile("(?is)\\b(SELECT\\s+.+?)(?:\\s+PARTITIONED\\s+BY|\\s+CLUSTERED\\s+BY|$)"); + + static final int DEFAULT_EMIT_QUEUE_CAPACITY = 1000; + static final int DEFAULT_EMIT_THREAD_COUNT = 1; + private static final int DISCARD_WARNING_INTERVAL = 1000; + + static final String UNKNOWN_QUERY_ID = "unknown-query-id"; + + private final ObjectMapper jsonMapper; + private final String namespace; + private final OpenLineageRequestLoggerProvider.TransportType transportType; + @Nullable + private final String transportUrl; + private final Set<String> excludedNativeQueryTypes; + @Nullable + private final HttpClient httpClient; + @Nullable + private final ExecutorService emitExecutor; + private final AtomicLong discardedEventCount = new AtomicLong(0); + + public OpenLineageRequestLogger( + ObjectMapper jsonMapper, + String namespace, + OpenLineageRequestLoggerProvider.TransportType transportType, + @Nullable String transportUrl, + Set<String> excludedNativeQueryTypes + ) + { + this(jsonMapper, namespace, transportType, transportUrl, excludedNativeQueryTypes, + DEFAULT_EMIT_QUEUE_CAPACITY, DEFAULT_EMIT_THREAD_COUNT, null); + } + + public OpenLineageRequestLogger( + ObjectMapper jsonMapper, + String namespace, + OpenLineageRequestLoggerProvider.TransportType transportType, + @Nullable String transportUrl, + Set<String> excludedNativeQueryTypes, + int emitQueueCapacity, + int emitThreadCount, + @Nullable HttpClient httpClient + ) + { + this.jsonMapper = jsonMapper; + this.namespace = namespace; + this.transportType = transportType; + this.transportUrl = transportUrl; + this.excludedNativeQueryTypes = excludedNativeQueryTypes; + if (transportType == OpenLineageRequestLoggerProvider.TransportType.HTTP && transportUrl == null) { + throw new IllegalStateException( + "druid.request.logging.transportUrl must be set when transportType=HTTP" + ); + } + if (transportType == OpenLineageRequestLoggerProvider.TransportType.HTTP) { + this.httpClient = httpClient; Review Comment: [P2] Direct HTTP logger construction leaves client null The public 5-argument constructor still delegates here with a null httpClient. After this update, HTTP mode stores that null instead of creating a default client, so direct construction with HTTP and a transportUrl enqueues events whose worker hits a NullPointerException at httpClient.execute and never posts lineage. Please keep default client creation when httpClient is null, or make the client-injection path non-public and route all HTTP construction through the provider. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
