Impala4.x源码阅读笔记（三）——Impala如何管理Iceberg表元数据

前言

本文为笔者个人阅读Apache Impala源码时的笔记，仅代表我个人对代码的理解，个人水平有限，文章可能存在理解错误、遗漏或者过时之处。如果有任何错误或者有更好的见解，欢迎指正。

上一篇文章Impala4.x源码阅读笔记（二）——Impala如何高效读取Iceberg表简单介绍了Iceberg表的基本情况和Impala是如何对其进行扫描的。这一篇则从元数据的角度对Impala如果管理Iceberg元数据进行一些简单的分析，这里的Iceberg元数据不是指Iceberg的那些元数据文件，那些是Iceberg API负责管理的，这里的元数据是指Iceberg表在Impala中的那些内存对象。

为了提升查询性能，Impala本身有一套比较复杂的元数据管理机制，这套机制以Catalogd服务进程为核心，实现了元数据在集群内的集中管理。在一个完整的Impala集群中，Catalogd服务进程主要担任了两个角色，首先是一个自动化的元数据缓存，它会负责缓存并自动同步Hive Metastore中的元数据，并将其广播给集群中其他负责处理查询的Coordinator节点，这使得Coordinator可以省去每次查询时和Hive Metastore（HMS）以及HDFS Namenode（NN）的交互，从而缩短了查询耗时。Catalogd的另一个角色是一个集中式的DDL执行者，其他Coordinator节点接收到的DDL最终都会以RPC的形式交由Catalogd进行执行，这样使得集群内部的元数据统一得到了保证。

Iceberg表作为一种表格式而非文件格式，其核心目标是高效且多功能地管理大量的数据文件，为了达成这一目标，Iceberg表的元数据相较于Hive表也更加复杂，从Impala支持Iceberg表的代码大部分都是元数据相关的也能看出这一点。关于Iceberg元数据管理的代码十分庞大，想要在一篇文章内全面地分析一遍是不太现实的，所以本文首先还是整体地、笼统地描述一下Iceberg表元数据的全貌，然后着重对元数据管理的两个关键环节——Iceberg表的加载和创建进行分析。

Iceberg表的相关接口

由于Impala本身元数据管理的特性和Iceberg表元数据的特殊性，在Impala支持Iceberg表各项功能的过程中定义了很多类型，我们首先看一下Iceberg相关类的整体UML图：

Iceberg UML

图中画出了Iceberg表在Impala中的主要相关接口和类以及其之间的实现或继承关系，其中绿色背景的就是与Iceberg表直接相关的，每个接口和类中都只列举了个别关键的成员变量和方法。在介绍具体的类之前我们先看一下其中的几个接口。

FeTable

首先是Impala中所有类型的表都要实现的接口FeTable，它定义了Impala Frontend与各类型表交互的一些基本操作，比如获取获取库表名、数据列列表、所有者等，其中还包括了可以获取org.apache.hadoop.hive.metastore.api.Table对象的getMetaStoreTable()方法。因为Impala对HMS是强依赖的，基本上所有元数据都来自HMS，为了与HMS进行元数据交互，Impala中所有类型的表中都包括一个HMS中表对象也就是org.apache.hadoop.hive.metastore.api.Table。它为Impala的表提供了基本的元数据，比如库表名、字段信息和表属性properties等。Iceberg表在Impala也不例外，需要在HMS注册了才能被Impala查询。当然Iceberg表本身并不一定依赖HMS，比如使用HadoopCatalog创建的Iceberg表只依赖一个像HDFS一样的支持原子重命名文件的文件系统而已。对于这种Iceberg表需要先在Impala中通过创建外表的方式在HMS进行注册才能被Impala元数据管理所接受。

FeFsTable

FeFsTable是Impala中所有基于文件系统的表类型都要实现的接口，它继承了FeTable接口，作用也是类似的。像存储在HDFS、S3这种常见的文件系统或存储服务上的表都属于FeFsTable，Iceberg表自然也是FeFsTable。在FeTable的基础之上，FeFsTable额外定义了许多和文件系统相关的方法，比如获取文件系统类型、表位置和文件系统对象FileSystem等。

FeIcebergTable

FeIcebergTable是Impala中Iceberg表类型都要实现的接口，继承了FeFsTable并额外定义了关于Iceberg的通用方法。其中有几个关键方法需要重点介绍：

getFeFsTable()，它会返回Iceberg对象内置的一个FeFsTable对象，这个对象会被用于将Iceberg表传递给Impala Backend。我们知道在Impala中Frontend负责制定执行计划、Backend负责执行，两者分别由Java和C++开发，之间主要通过Thrift结构体传递数据，这些数据也包括了查询的执行计划。而Iceberg表在执行期间与普通HDFS表实际上并没有显著差别，可以说都只是一系列规划好的数据文件而已。因此，为了复用Backend中现有的HDFS表扫描代码，Iceberg表对象都内置了一张普通HDFS表对象，在序列化为Thrift结构体传递给Backend时就使用这个内置的FeFsTable对象的相关方法将自身“转变”为HDFS表。
getIcebergApiTable()，它会返回Iceberg表对象对应的Iceberg API中的表对象org.apache.iceberg.Table，这是Iceberg API中的表示Iceberg表的接口，它提供了Iceberg表的许多重要API，比如获取快照、Schema和扫描计划。依靠这些接口，Impala可以进行Iceberg表的时间旅行查询、模式演进、谓词下推和获取数据文件列表等操作。
getIcebergCatalog()，它会返回Iceberg表的Catalog类型，目前Impala支持的Iceberg Catalog类型有HadoopTables、HadoopCatalog、HiveCatalog和Catalogs。Iceberg的Catalog是用于追踪Iceberg表的，它主要负责储存Iceberg表最近元数据文件的位置，可以说是Iceberg表元数据的元数据。换句话说如果说Iceberg表是管理一系列数据文件并告诉我们数据文件在哪里，那么Iceberg Catalog的作用就是管理一系列Iceberg表并告诉我们Iceberg表在哪里。如果Iceberg表的最近元数据位置也直接储存在文件系统的一个文件中，则对应HadoopTables。如果文件系统中有一个专门的Catalog目录，Iceberg表的元数据位置由其负责管理，则对应HadoopCatalog。如果使用HMS储存Iceberg表的最近元数据位置，则对应HiveCatalog。而Catalogs接口相当于一种复合的自动Catalog，它依赖配置文件和表属性自动识别Iceberg表的Catalog类型。

接口FeIcebergTable可以说是Iceberg表在Impala中的关键抽象，从图中也可以看到许多Iceberg表的相关类实现了该接口。

Iceberg表的相关类

介绍完了相关接口之后，我们接下来继续看看Iceberg相关的类。从图中可以看到与Iceberg直接相关的表类型就有足足七种，当然其中除了IcebergTable和LocalIcebergTable这两个真正表示实际存在的Iceberg表的“正经”表类型外，其他的都可以算是为了支持各种Iceberg特性而抽象出来的功能性的工具类。接下来我们逐个介绍。

IcebergTable

IcebergTable是Iceberg表在Impala元数据管理中的代理类之一，每个对象都是对应了一张实际存在的Iceberg表。IcebergTable实现了FeIcebergTable接口并继承了Table类。Table类是Impala中所有表类的主要父类之一（另一个是LocalTable），它是一个抽象类，实现了FeTable接口，它定义了所有表共有的一些成员变量，如库对象、表名、所有者、表锁和数据列容器等等，它还定义了表对象共有的一些成员方法，其中最重要的就是实现表加载的抽象方法load()以及Coordinator接收到Catalogd服务广播的元数据Thrift结构体后从Thrift结构体加载元数据的loadFromThrift(TTable)方法。IcebergTable作为Table的子类，实现了自己的load()方法来加载Iceberg表，除此之外还包括一些特有的成员，如前文提到的内置的HDFS表对象hdfsTable_、Iceberg API表对象icebergApiTable_和从Iceberg元数据加载Schema的方法loadSchemaFromIceberg()等等。

LocalIcebergTable

LocalIcebergTable可以理解为IcebergTable的Local版本，它只在Coordinator的Local Catalog模式下使用，而IcebergTable会在Catalogd和Coordinator的传统Catalog模式下使用，LocalIcebergTable在Coordinator的作用和IcebergTable基本是一致的，可以说是更加轻量化的IcebergTable。Local Catalog模式是为了解决传统Catalog模式的一些缺点而设计的，它支持更细粒度的元数据缓存并能在启动时按需加载元数据，提升了Coordinator的启动速度并减少了内存消耗。LocalIcebergTable同样实现了FeIcebergTable接口，但是继承的是LocalTable类，而不是Table类。LocalTable类也是抽象类，是Table类的Local版本，其成员LocalDb这是Db类的Local版本。如同IcebergTable一样，LocalIcebergTable也内置了一张HDFS表对象，不过不再是HdfsTable类了，而是其Local版本的LocalFsTable，这些Local类都是只在Coordinator的Local Catalog模式下使用的，和非Local版本一一对应。

IcebergPositionDeleteTable

IcebergPositionDeleteTable是用于Iceberg MOR的虚拟表，在上一篇文章中其实已经登场过了，它只在制定Iceberg的Position Delete扫描计划中会被使用到，用来将Iceberg表的Delete File组织为一张虚拟表，这样才能使用Impala的ScanNode进行扫描，具体的使用过程可以参考上一篇文章Impala4.x源码阅读笔记（二）——Impala如何高效读取Iceberg表。IcebergPositionDeleteTable同样实现了FeIcebergTable接口，不过它继承的是表示虚拟表的抽象类VirtualTable，虚拟表不是实际存在的表，而是为了实现某些特定功能而虚拟出来的表，它往往会根据需要而添加一些虚拟列，可以将非表形式的数据以表的形式进行处理。

IcebergMetadataTable

Iceberg API提供了一系列专门的元数据表来查询Iceberg表的元数据，可通过其MetadataTableUtils类来创建各种类型的Iceberg元数据表，如ManifestEntriesTable、FilesTable和SnapshotsTable等。这些元数据表基于基本的Iceberg表创建，有各自的Schema，用于查询该表的各种元数据。IcebergMetadataTable就是Impala为了对接这些Iceberg元数据表而定义的类，它是另外一个继承了VirtualTable的类，不过它并没有实现FeIcebergTable接口，因为它不是通常的Iceberg表。它可以根据一个FeIcebergTable对象和元数据表类型字符串来创建，利用MetadataTableUtils来对接Iceberg元数据表获取Schema并依此填充自身作为VirtualTable的虚拟列，执行时IcebergMetadataTable由执行引擎这边的专门的IcebergMetadataScanNode负责扫描，当然由于执行引擎是C++编写的，所以实际扫描时还是需要通过JNI调用Iceberg API来完成。

IcebergCtasTarget

IcebergCtasTarget是用于CTAS（Create Table As Select）语句的临时目标表类型，它继承了CtasTargetTable类并实现了FeIcebergTable接口，不过它也不是实际存在的表，只是用于CTAS的分析过程。Impala分析CTAS语句时会将其分解为CREATE语句和INSERT语句，然后根据CREATE语句先创建临时目标表，再结合临时目标表来分析INSERT语句。如果分析过程顺利完成才会真正创建目标表。对于Iceberg表来说，临时目标表只是分析使用的，不应该通过Iceberg API实际创建它，所以需要IcebergCtasTarget来充当这一角色。IcebergCtasTarget实现了FeIcebergTable，但是并不会通过Iceberg API实际创建一张Iceberg表。

ForwardingFelcebergTable

ForwardingFelcebergTable一个用于FeIcebergTable的转发类，也并非什么实际存在的表，只是一种使用组合代替继承的编程技巧，通过ForwardingFelcebergTable可以在不继承基类的前提下将不需要重写的方法委托给基类FeIcebergTable。这个类会在IcebergTimeTravelTable中使用，避免IcebergTimeTravelTable继承IcebergTable、LocalIcebergTable等类。

IcebergTimeTravelTable

IcebergTimeTravelTable表示进行时间旅行的Iceberg表，由于Iceberg表时间旅行和模式演进的特性，在不同的时间点Iceberg表可能有不同的Schema，因此对于进行时间旅行的Iceberg表我们需要根据时间或版本重新加载Schema，为了避免复制或破坏原始的Iceberg元数据，Impala通过IcebergTimeTravelTable来实现时间旅行的Iceberg表。IcebergTimeTravelTable没有继承FeIcebergTable而是继承了ForwardingFelcebergTable，通过ForwardingFelcebergTable嵌入对原始Iceberg表的引用并在此基础之上实现readSchema()加载自己的Schema，而那些未涉及时间旅行的方法都可以通过ForwardingFelcebergTable委托给原始Iceberg表类的同名方法。

至此Iceberg表在Impala中的相关类就介绍完了，可以发现除了IcebergTable和LocalIcebergTable可以真正称得上是Iceberg表的元数据之外，其他的类都是为了实现Iceberg表的各种功能而定义工具类，这些繁多的类看起来复杂，实际上本身代码量并不多，不如说正是因为定义了这些类才使得Iceberg元数据更好地融入Impala的元数据体系，也使得Impala在支持Iceberg的过程中可以大量复用现有的、可靠的、高性能的代码，反而减少了开发工作量。

Iceberg表的加载

接下来我们分析一下Iceberg表在Impala中是如何加载的，所谓表的加载实际上就是Impala根据HMS的元数据对象创建自己的元数据对象的过程，对于Iceberg也是一样的，不过Iceberg还有很大一部分元数据以文件的形式存在，需要Iceberg API处理。不过在调用IcebergTable的load()方法之前我们需要先知道它是一张Iceberg表，这一判断由其静态方法isIcebergTable()完成：

  public static boolean isIcebergTable(org.apache.hadoop.hive.metastore.api.Table msTbl) {// 从HMS元数据获取InputFormat，如果是org.apache.iceberg.mr.hive.HiveIcebergInputFormat// 则HdfsFileFormat会是HdfsFileFormat.ICEBERGString inputFormat = msTbl.getSd().getInputFormat();HdfsFileFormat hdfsFileFormat = inputFormat != null ?HdfsFileFormat.fromHdfsInputFormatClass(inputFormat, null) :null;// 如果表属性中的storage_handler值为org.apache.iceberg.mr.hive.HiveIcebergStorageHandler// 或者HdfsFileFormat为HdfsFileFormat.ICEBERG// 或者table_type值为ICEBERG，则会认为这是一张Iceberg表return isIcebergStorageHandler(msTbl.getParameters().get(KEY_STORAGE_HANDLER)) ||hdfsFileFormat == HdfsFileFormat.ICEBERG ||(hdfsFileFormat == null &&"ICEBERG".equals(msTbl.getParameters().get("table_type")));}

根据表属性判断是一张Iceberg表之后，就可以使用IcebergTable的load()方法加载元数据了：

@Override
public void load(boolean reuseMetadata, IMetaStoreClient msClient,org.apache.hadoop.hive.metastore.api.Table msTbl, String reason)throws TableLoadingException {... // 省略一些非关键代码// IcebergUtil.loadTable()方法会通过Iceberg API加载Iceberg表元数据，返回一个Iceberg的Table对象icebergApiTable_ = IcebergUtil.loadTable(this);catalogSnapshotId_ = FeIcebergTable.super.snapshotId();// loadSchemaFromIceberg()方法会将Iceberg Schema转换为Hive Schema并设置到HMS的Table对象msTable_中// 同时还会将Iceberg Schema转换为Impala的Column并添加到自身的列容器colsByPos_和colsByName_中// 这些转换的过程实际上就是遍历Iceberg Schema的每个字段，创建对应类型的Hive或Impala类型// 此外还有添加虚拟列、加载分区Spec等操作loadSchemaFromIceberg();// 然后是一些表属性的设置icebergFileFormat_ = IcebergUtil.getIcebergFileFormat(msTbl);icebergParquetCompressionCodec_ = Utils.getIcebergParquetCompressionCodec(msTbl);icebergParquetRowGroupSize_ = Utils.getIcebergParquetRowGroupSize(msTbl);icebergParquetPlainPageSize_ = Utils.getIcebergParquetPlainPageSize(msTbl);icebergParquetDictPageSize_ = Utils.getIcebergParquetDictPageSize(msTbl);// 通过IcebergUtil.getIcebergFiles()方法可以获取Iceberg表的数据文件集合// 这个方法还支持传入谓词列表和时间旅行描述来进行谓词下推和时间旅行，得到对应的文件集合// 这里要获取最新快照的全部数据文件来缓存，所以传入空列表和空指针GroupedContentFiles icebergFiles = IcebergUtil.getIcebergFiles(this,new ArrayList<>(), /*timeTravelSpec=*/null);// 最后我们还需要加载Iceberg内置的Hdfs表hdfsTable_.setIcebergFiles(icebergFiles);hdfsTable_.setCanDataBeOutsideOfTableLocation(!Utils.requiresDataFilesInTableLocation(this));hdfsTable_.load(reuseMetadata, msClient, msTable_, reason);... // 省略一些非关键代码
}

可以看到加载过程中调用了许多其他方法，不过其中最关键的还是用来加载Iceberg API中的Table对象的IcebergUtil.loadTable()方法：

public static Table loadTable(FeIcebergTable feTable) throws IcebergTableLoadingException {// 调用下面的重载方法return loadTable(feTable.getIcebergCatalog(), getIcebergTableIdentifier(feTable),feTable.getIcebergCatalogLocation(), feTable.getMetaStoreTable().getParameters());
}public static Table loadTable(TIcebergCatalog catalog, TableIdentifier tableId,String location, Map<String, String> tableProps) throws IcebergTableLoadingException {...// 根据Catalog类型获取对应的IcebergCatalog实例，然后使用该Catalog实例的loadTable方法加载Iceberg表IcebergCatalog cat = getIcebergCatalog(catalog, location);return cat.loadTable(tableId, location, tableProps);
}public static IcebergCatalog getIcebergCatalog(TIcebergCatalog catalog, String location)throws ImpalaRuntimeException {switch (catalog) {// 正如前文所述，Impala目前支持四种Iceberg Catalog，它们在Impala中都对应了各自的单例对象// 这些对象都实现了Impala的IcebergCatalog接口，提供诸如createTable()/loadTable()/dropTable()等方法// 这些IcebergCatalog类封装了对应的Iceberg API，比如IcebergHiveCatalog封装了Iceberg包的HiveCatalog// 它们的loadTable()实际上也就是调用了对应的Iceberg API中Catalog的loadTable()方法case HADOOP_TABLES: return IcebergHadoopTables.getInstance();case HIVE_CATALOG: return IcebergHiveCatalog.getInstance();case HADOOP_CATALOG: return new IcebergHadoopCatalog(location);case CATALOGS: return IcebergCatalogs.getInstance();default: throw new ImpalaRuntimeException("Unexpected catalog type: " + catalog);}
}

Iceberg表的加载过程虽然步骤很多但还是比较清晰的，每部分逻辑都封装为了特定的方法来完成，从方法名也能大致了解其作用。

Iceberg表的创建

除了表的加载之外，表的创建也是元数据另外一个源头，在Impala中DDL由Coordinator解析&分析为特定类型的参数集合，然后通过RPC远程调用Catalogd进程的方法来执行。建表的过程也不例外，在Catalogd进程的Catalog操作执行类CatalogOpExecutor中由方法createTable()完成：

private boolean createTable(TCreateTableParams params, TDdlExecResponse response,EventSequence catalogTimeline, boolean syncDdl, boolean wantMinimalResult)throws ImpalaException {... // 省略一些非关键代码// 根据建表参数params先创建一个基本的HMS Table对象，后续建表过程主要就依赖这个对象了org.apache.hadoop.hive.metastore.api.Table tbl = createMetaStoreTable(params);LOG.trace("Creating table {}", tableName);if (KuduTable.isKuduTable(tbl)) {// 创建Kudu表的分支return createKuduTable(tbl, params, wantMinimalResult, response, catalogTimeline);} else if (IcebergTable.isIcebergTable(tbl)) {// 创建Iceberg表的分支，调用更具体的createIcebergTable()方法来进行return createIcebergTable(tbl, wantMinimalResult, response, catalogTimeline,params.if_not_exists, params.getColumns(), params.getPartition_spec(),params.getTable_properties(), params.getComment());}... // 省略一些非关键代码
}

我们接着看createIcebergTable()方法：

private boolean createIcebergTable(org.apache.hadoop.hive.metastore.api.Table newTable,boolean wantMinimalResult, TDdlExecResponse response, EventSequence catalogTimeline,boolean ifNotExists, List<TColumn> columns, TIcebergPartitionSpec partitionSpec,Map<String, String> tableProperties, String tblComment) throws ImpalaException {... // 省略部分代码// 首先获取Iceberg表的Catalog类型，这直接决定了Iceberg表的创建方法TIcebergCatalog catalog = IcebergUtil.getTIcebergCatalog(newTable);String location = newTable.getSd().getLocation();// 如果用户在通过Impala创建一张全新的Iceberg表，也就是同步表，则需要先通过Iceberg API创建一张Iceberg表// 所谓同步表也就是非外表或设置了Purge的外表，这两种情况下我们都期望目标Iceberg表还不存在if (IcebergTable.isSynchronizedTable(newTable)) {// 在使用Iceberg API创建Iceberg表前先需要明确表的创建位置// 如果SQL中没有指定表位置，我们需要根据Catalog类型指定一个表位置if (location == null) {if (catalog == TIcebergCatalog.HADOOP_CATALOG) {// 使用Hadoop Catalog时，建表时应当通过表属性iceberg.catalog_location明确指定Hadoop Catalog的位置location = IcebergUtil.getIcebergCatalogLocation(newTable);} else {// 使用其他Catalog时，使用HMS的API为新表生成一个位置location = MetastoreShim.getPathForNewTable(msClient.getHiveClient().getDatabase(newTable.getDbName()),newTable);}}// 通过IcebergCatalogOpExecutor调用Iceberg API创建一张Iceberg表// IcebergCatalogOpExecutor.createTable()方法会创建Iceberg API的Schema、PartitionSpec对象// 然后根据Catalog类型获取对应的IcebergCatalog实例，然后使用其createTable方法创建Iceberg表// 这一过程与加载Iceberg表时的操作还有些类似String tableLoc = IcebergCatalogOpExecutor.createTable(catalog,IcebergUtil.getIcebergTableIdentifier(newTable), location, columns,partitionSpec, newTable.getOwner(), tableProperties).location();newTable.getSd().setLocation(tableLoc);catalogTimeline.markEvent(CREATED_ICEBERG_TABLE + catalog.name());} else {// 如果不是在创建同步表，那我们期望Iceberg Catalog中已经存在我们需要的Iceberg表了// 此时Impala创建Iceberg表的行为更类似于加载现有的Iceberg表并将其注册到HMS中// 首先同样是先需要得到目标表的Catalog类型TIcebergCatalog underlyingCatalog = IcebergUtil.getUnderlyingCatalog(newTable);String locationToLoadFrom;// 然后根据Catalog类型确定我们从何处加载Iceberg表if (underlyingCatalog == TIcebergCatalog.HADOOP_TABLES) {// 对于Hadoop Tables，直接从建表SQL指定的locatin加载表if (location == null) {addSummary(response,"Location is necessary for external iceberg table.");return false;}locationToLoadFrom = location;} else {// 对于Hadoop Catalog，依然从表属性iceberg.catalog_location获取Hadoop Catalog的位置locationToLoadFrom = IcebergUtil.getIcebergCatalogLocation(newTable);}// 然后通过上文介绍过的IcebergUtil.loadTable()将Iceberg表加载起来TableIdentifier identifier = IcebergUtil.getIcebergTableIdentifier(newTable);org.apache.iceberg.Table iceTable = IcebergUtil.loadTable(catalog, identifier, locationToLoadFrom, newTable.getParameters());... // 省略部分代码
}

可以发现Iceberg表的创建过程和加载过程还是有些类似的，主要都是根据Catalog类型来调用对应的Iceberg Catalog API来实现Iceberg表本身的元数据操作，然后Impala自身负责完成Schema转换、分区转换和配置填充等工作。

总结

这篇文章主要是从Iceberg表的相关接口和类以及表的加载和创建两个方面分析了Iceberg表元数据在Impala中是如何管理的，总的来说为了在Impala中方便且高效地实现Iceberg表的各种功能，代码中定义了许多相关的类，虽然看起来比较复杂，但是每个类的功能用途都很明确。而在表的加载和创建方面，Impala也支持了多种Iceberg Catalog，并能和现有的基于HMS的元数据缓存框架结合起来，这使得用户不用操心诸如元数据同步等问题，使用起来还是比较丝滑的。限于文章篇幅，实际上代码中还有许多内容无法详细展开分析，而且Impala社区目前也在重点开发Iceberg相关的特性，代码变化也比较快，有兴趣的同学也可以直接关注Impala的Github仓库和Jira关注社区最新进展。