Why using intput.schema() in keyCalculationExtension?

[zhuliquan]

I notice that schema in func output_scheame is not compatible with func schema in keyCalculationExtension.

in output_schemme, using self.input.schema() as output.

arroyo/crates/arroyo-planner/src/extension/key_calculation.rs

Lines 112 to 115 in 4f8dc2e

	fn output_schema(&self) -> ArroyoSchema {
	let arrow_schema = Arc::new(self.input.schema().as_ref().into());
	ArroyoSchema::from_schema_keys(arrow_schema, self.keys.clone()).unwrap()
	}

but in schema, using self.schema as output.

arroyo/crates/arroyo-planner/src/extension/key_calculation.rs

Lines 127 to 130 in 4f8dc2e

	fn schema(&self) -> &DFSchemaRef {
	&self.schema
	}

I notice that self.schema is not equal to self.input.schema()

arroyo/crates/arroyo-planner/src/extension/key_calculation.rs

Lines 39 to 58 in 4f8dc2e

	let output_fields: Vec<_> = fields_with_qualifiers(input.schema())
	.into_iter()
	.enumerate()
	.filter_map(|(index, field)| {
	if !keys.contains(&index) {
	Some(field.clone())
	} else {
	None
	}
	})
	.collect();
	let schema =
	schema_from_df_fields_with_metadata(&output_fields, input.schema().metadata().clone())
	.unwrap();
	Self {
	name: Some(name),
	input,
	keys,
	schema: Arc::new(schema),
	}