Hazelcast Jet JdbcSqlConnector Arayüzü - Stateless Çalışır

Giriş

Şu satırı dahil ederiz

import com.hazelcast.jet.sql.impl.connector.jdbc.JdbcSqlConnector;

Not : Aslında bu sınıf biraz kötü bir tasarım çünkü Single Responsibility kuralını ihlal ediyor. Çünkü bu sınıf

1. Hem mapping yaratılırken veri tabanındaki tabloyu keşfediyor

2. Hem Hazelcast SQL cümlelerini çalıştırıyor

SqlConnector arayüzünden kalıtır. Mapping type olarak JDBC kullanan mapping'lerle ilgilenir Kodu şöyle

public class JdbcSqlConnector implements SqlConnector {
  public static final String TYPE_NAME = "JDBC";
  ...
}

- Hazelcast SQL cümlelerini gerçek veri tabanına gönderilecek SQL cümlesi haline getirir. 

- Gerçek SQL cümlesi haline getirme işini builder sınıflar yapar. Bu sınıflar şöyle

• DeleteQueryBuilder
• InsertQueryBuilder 
• SelectQueryBuilder, 
• UpdateQueryBuilder

Gerçek SQL cümlesini şu metodlar yerine getirir. Bu sınıflar yeni bir processor nesnesi yaratıp çalışması için Jet Engine'e verirler.

• deleteProcessor()
• insertProcessor()
• fullScanReade()
• updateProcessor()
• sinkProcessor()

createTable metodu

İmzası şöyle

public Table createTable(
            NodeEngine nodeEngine,
            String schemaName,
            String mappingName,
            String[] externalName,
            String dataConnectionName,
            Map<String, String> options,
            List<MappingField> resolvedFields)

TableResolverImpl tarafından çağrılır. Tanımlı olan her mapping için bir tane JdbcTable yaratır.

deleteProcessor metodu

İmzası şöyle. DeleteProcessorSupplier aracılığıyla WriteJdbcP yaratır

public Vertex deleteProcessor(DagBuildContext context)

fullScanReader metodu

İmzası şöyle. SelectProcessorSupplier aracılığıyla  ReadJdbcP yaratır

public Vertex fullScanReader(
  DagBuildContext context,
  HazelcastRexNode predicate,
  List<HazelcastRexNode> projection,
  FunctionEx<ExpressionEvalContext, EventTimePolicy<JetSqlRow>> eventTimePolicyProvider
)

insertProcessor metodu

İmzası şöyle. InsertProcessorSupplier aracılığıyla WriteJdbcP yaratır

public VertexWithInputConfig insertProcessor(DagBuildContext context)

InsertQueryBuilder nesnesini kullanarak JdbcTable nesnesinde belirtilen tüm sütunları kullanan bir INSERT cümlesi üretir. Bu cümle bir InsertProcessorSupplier nesnesine geçilir ve InsertProcessorSupplier da bir Vertex nesnesine eklenir.

updateProcessor metodu

İmzası şöyle. UpdateProcessorSupplier aracılığıyla  WriteJdbcP yaratır

public Vertex updateProcessor(
            DagBuildContext context,
            List<String> fieldNames,
            List<HazelcastRexNode> expressions)

resolveAndValidateFields metodu

İmzası şöyle

public List<MappingField> resolveAndValidateFields(
  NodeEngine nodeEngine,
  Map<String, String> options,
  List<MappingField> userFields,
  String[] externalName,
  String dataConnectionName)

TableResolverImpl tarafından "CREATE MAPPING ..." SQL cümlesi için çağrılır. Belirtilen mapping için mapping yaratılırken external name ile belirtilen tabloya ait sütunları keşfeder. Çağrı sırası kabaca şöyle

SqlService.execute(String sql, Object... arguments)
SqlServiceImpl.execute(SqlStatement statement)
SqlServiceImpl.execute(SqlStatement statement, SqlSecurityContext securityContext)
...
SqlServiceImpl.execute(SqlStatement statement, SqlSecurityContext securityContext,
            QueryId queryId,
            boolean skipStats
)
SqlPlanImpl.execute(QueryId queryId, List<Object> arguments, long timeout)
PlanExecutor.execute(CreateMappingPlan plan)
TableResolverImpl.createMapping(Mapping mapping, boolean replace, boolean ifNotExists)
TableResolverImpl.resolveMapping(Mapping mapping)

Eğer userFields belirtilmişse sadece bu sütunla keşfedilir. Belirtilmemişse tüm sütunlar keşfedilir. 

Keşif sırasında tabloya ait bilgiler DbField denilen geçici bir nesne içinde saklanır. Daha sonra DbField nesnesi, MappingField nesnesine çevrilir. Metod bir List<MappingField> döndürür.

resolveDialect metodu

Bir tane JDBC DataSource nesnesi yaratır. Bu DataSource nesnesinden JDBC Connection nesnesi alır. Connection nesnesinden de DatabaseMetaData nesnesi alır. Sonra Apache Calcite kütüphanesine bu DatabaseMetaData nesnesi için bir SqlDialect nesnesi üretmesini söyler.